diff --git a/common/src/main/java/org/apache/sedona/common/S2Geography/WKBGeography.java b/common/src/main/java/org/apache/sedona/common/S2Geography/WKBGeography.java
index 8ef77d973ca..c77297f03fa 100644
--- a/common/src/main/java/org/apache/sedona/common/S2Geography/WKBGeography.java
+++ b/common/src/main/java/org/apache/sedona/common/S2Geography/WKBGeography.java
@@ -26,10 +26,12 @@
 import com.google.common.geometry.S2Shape;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.util.Arrays;
 import java.util.List;
 import org.locationtech.jts.geom.Geometry;
 import org.locationtech.jts.geom.PrecisionModel;
 import org.locationtech.jts.io.ParseException;
+import org.locationtech.jts.io.WKBConstants;
 
 /**
  * A Geography implementation that stores WKB bytes as the primary representation, with lazy-parsed
@@ -80,6 +82,14 @@ private WKBGeography(byte[] wkbBytes, int srid) {
    */
   public static WKBGeography fromWKB(byte[] wkb, int srid) {
     WKBGeography geog = new WKBGeography(wkb, srid);
+    // Sedona versions before 1.9.1 could emit an empty LineString as only its 5-byte WKB header
+    // (or 9 bytes when an EWKB SRID was present), without the required zero coordinate count.
+    // Normalize that legacy representation at the boundary so it never escapes as malformed WKB.
+    if (wkb.length >= 5
+        && geog.wkbBaseType() == WKBConstants.wkbLineString
+        && wkb.length == geog.wkbPayloadOffset()) {
+      geog = new WKBGeography(Arrays.copyOf(wkb, wkb.length + Integer.BYTES), srid);
+    }
     if (eagerShapeIndex) {
       geog.getShapeIndexGeography();
     }
@@ -245,6 +255,27 @@ public boolean isPoint() {
     return wkbBaseType() == 1;
   }
 
+  /**
+   * Returns true if the top-level WKB is empty without constructing JTS or S2 objects. For
+   * non-point WKB, the first payload integer is the element/ring/vertex count.
+   */
+  public boolean isEmpty() {
+    int type = wkbBaseType();
+    int payloadOffset = wkbPayloadOffset();
+    if (type == 1) {
+      if (wkbBytes.length < payloadOffset + 2 * Double.BYTES) return false;
+      return getPointX() == null;
+    }
+    if (type >= 2 && type <= 7) {
+      if (wkbBytes.length < payloadOffset + Integer.BYTES) return false;
+      boolean le = (wkbBytes[0] == 0x01);
+      ByteBuffer bb =
+          ByteBuffer.wrap(wkbBytes).order(le ? ByteOrder.LITTLE_ENDIAN : ByteOrder.BIG_ENDIAN);
+      return bb.getInt(payloadOffset) == 0;
+    }
+    return false;
+  }
+
   /** Returns the X coordinate directly from Point WKB, or null for non-points and empty points. */
   public Double getPointX() {
     return getPointOrdinate(0);
diff --git a/common/src/main/java/org/apache/sedona/common/geography/Functions.java b/common/src/main/java/org/apache/sedona/common/geography/Functions.java
index f495d86c723..6c246752c43 100644
--- a/common/src/main/java/org/apache/sedona/common/geography/Functions.java
+++ b/common/src/main/java/org/apache/sedona/common/geography/Functions.java
@@ -20,11 +20,14 @@
 
 import com.google.common.geometry.*;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import org.apache.sedona.common.S2Geography.*;
 import org.apache.sedona.common.sphere.Haversine;
 import org.locationtech.jts.geom.Coordinate;
 import org.locationtech.jts.geom.Geometry;
+import org.locationtech.jts.geom.GeometryCollection;
 import org.locationtech.jts.geom.GeometryFactory;
 import org.locationtech.jts.geom.LineString;
 import org.locationtech.jts.geom.MultiPoint;
@@ -35,6 +38,8 @@
 public class Functions {
 
   private static final double EPSILON = 1e-9;
+  private static final S1Angle PROJECT_PERPENDICULAR_ERROR =
+      S1Angle.radians((2.0 + 2.0 / Math.sqrt(3.0)) * S2.DBL_ERROR).add(S2.ROBUST_CROSS_PROD_ERROR);
   // S2 expands conservative rectangle bounds by a few floating-point ulps. Snap only those tiny
   // expansions back to a source ordinate; genuine great-circle extrema remain unchanged.
   private static final double BOUND_ORDINATE_SNAP_TOLERANCE_DEGREES = 1e-12;
@@ -267,6 +272,280 @@ public static Double y(Geography g) {
     return org.apache.sedona.common.Functions.y(toJTS(g));
   }
 
+  /**
+   * Returns the smallest convex region containing {@code g} on the sphere.
+   *
+   * <p>The result follows the usual convex-hull dimensionality rules: one unique vertex produces a
+   * point, two or more collinear vertices produce a line, and non-collinear vertices produce a
+   * polygon whose edges are geodesics. Empty inputs preserve their input type.
+   *
+   * @throws UnsupportedOperationException when the hull is the full sphere, which cannot be
+   *     represented by OGC WKB
+   */
+  public static Geography convexHull(Geography g) {
+    if (g == null) return null;
+    if (g instanceof WKBGeography && ((WKBGeography) g).isEmpty()) return g;
+    if (!(g instanceof WKBGeography) && g.numShapes() == 0) return g;
+
+    Geography typed = (g instanceof WKBGeography) ? ((WKBGeography) g).getS2Geography() : g;
+    S2ConvexHullQuery query = new S2ConvexHullQuery();
+    List<S2Point> vertices = new ArrayList<>();
+    Geometry sourceGeometry = null;
+    boolean needsJtsFallback = addPointAndLineVertices(typed, query, vertices);
+    if (needsJtsFallback) {
+      sourceGeometry = toJTS(g);
+      if (sourceGeometry != null) {
+        addJtsPointAndLineVertices(sourceGeometry, query, vertices);
+      }
+    }
+    boolean hasPolygon = addPolygonRegions(typed, query, vertices);
+    if (vertices.isEmpty() && !hasPolygon) return g;
+
+    S2Loop loop = query.getConvexHull();
+    if (loop.isFull()) {
+      throw new UnsupportedOperationException(
+          "ST_ConvexHull produced the full sphere, which cannot be represented as OGC WKB");
+    }
+
+    S2Point[] degenerateHull = vertices.isEmpty() ? null : getDegenerateHull(vertices);
+    if (degenerateHull != null) {
+      if (sourceGeometry == null) sourceGeometry = toJTS(g);
+      return createDegenerateHull(sourceGeometry, degenerateHull, g.getSRID());
+    }
+
+    if (sourceGeometry == null) sourceGeometry = toJTS(g);
+    return createPolygonHull(sourceGeometry, loop, g.getSRID());
+  }
+
+  /**
+   * Collects geographies into the corresponding multi-geography or geography collection. Null
+   * elements are ignored, member order and duplicates are retained, and the first non-null input
+   * supplies the output SRID.
+   */
+  public static Geography createMultiGeography(Geography[] geographies) {
+    List<Geometry> geometries = new ArrayList<>();
+    Integer srid = null;
+    for (Geography geography : geographies) {
+      if (geography == null) continue;
+      if (srid == null) srid = geography.getSRID();
+      Geometry geometry = toJTS(geography);
+      if (geometry != null) geometries.add(geometry);
+    }
+
+    Geometry result =
+        org.apache.sedona.common.Functions.createMultiGeometry(geometries.toArray(new Geometry[0]));
+    if (srid != null) result.setSRID(srid);
+    return WKBGeography.fromJTS(result);
+  }
+
+  /**
+   * Adds point and polyline vertices from the S2 representation.
+   *
+   * @return whether an empty S2 polyline needs a JTS fallback to recover coincident source vertices
+   *     that S2Builder collapsed
+   */
+  private static boolean addPointAndLineVertices(
+      Geography geography, S2ConvexHullQuery query, List<S2Point> vertices) {
+    if (geography instanceof PointGeography) {
+      for (S2Point point : ((PointGeography) geography).getPoints()) {
+        query.addPoint(point);
+        vertices.add(point);
+      }
+      return false;
+    }
+    if (geography instanceof PolylineGeography) {
+      List<S2Polyline> polylines = ((PolylineGeography) geography).getPolylines();
+      boolean needsJtsFallback = polylines.isEmpty();
+      for (S2Polyline polyline : polylines) {
+        if (polyline.numVertices() < 2) {
+          needsJtsFallback = true;
+        } else {
+          query.addPolyline(polyline);
+          vertices.addAll(polyline.vertices());
+        }
+      }
+      return needsJtsFallback;
+    }
+    if (geography instanceof GeographyCollection) {
+      boolean needsJtsFallback = false;
+      for (Geography feature : ((GeographyCollection) geography).getFeatures()) {
+        needsJtsFallback |= addPointAndLineVertices(feature, query, vertices);
+      }
+      return needsJtsFallback;
+    }
+    return false;
+  }
+
+  private static void addJtsPointAndLineVertices(
+      Geometry geometry, S2ConvexHullQuery query, List<S2Point> vertices) {
+    if (geometry instanceof Polygon) return;
+    if (geometry instanceof Point || geometry instanceof LineString) {
+      for (org.locationtech.jts.geom.Coordinate coordinate : geometry.getCoordinates()) {
+        S2Point vertex = S2LatLng.fromDegrees(coordinate.getY(), coordinate.getX()).toPoint();
+        query.addPoint(vertex);
+        vertices.add(vertex);
+      }
+      return;
+    }
+    if (geometry instanceof GeometryCollection) {
+      for (int i = 0; i < geometry.getNumGeometries(); i++) {
+        addJtsPointAndLineVertices(geometry.getGeometryN(i), query, vertices);
+      }
+    }
+  }
+
+  /**
+   * Adds polygon regions to the hull query. Polygons must be added as regions rather than as vertex
+   * sets so the query respects the S2 representation's resolved interior. Public Geography readers
+   * normalize shells to at most one hemisphere, while directly constructed S2 Geography values can
+   * still represent a larger directed region whose hull is the full sphere.
+   *
+   * @return whether a non-empty polygon was added
+   */
+  private static boolean addPolygonRegions(
+      Geography geography, S2ConvexHullQuery query, List<S2Point> vertices) {
+    if (geography instanceof PolygonGeography) {
+      S2Polygon polygon = ((PolygonGeography) geography).polygon;
+      if (polygon.isEmpty()) return false;
+      query.addPolygon(polygon);
+      for (int i = 0; i < polygon.numLoops(); i++) {
+        S2Loop loop = polygon.loop(i);
+        if (loop.depth() == 0 && !loop.isEmptyOrFull()) {
+          vertices.addAll(loop.vertices());
+        }
+      }
+      return true;
+    }
+    if (geography instanceof GeographyCollection) {
+      boolean hasPolygon = false;
+      for (Geography feature : ((GeographyCollection) geography).getFeatures()) {
+        hasPolygon |= addPolygonRegions(feature, query, vertices);
+      }
+      return hasPolygon;
+    }
+    return false;
+  }
+
+  /**
+   * Returns either one point for a point hull, two endpoints for a collinear hull, or {@code null}
+   * for a polygonal hull. The two-pass farthest-point search is linear and finds the endpoints of
+   * any set contained by one geodesic segment.
+   */
+  private static S2Point[] getDegenerateHull(List<S2Point> vertices) {
+    S2Point firstEndpoint = farthestPoint(vertices.get(0), vertices);
+    S2Point secondEndpoint = farthestPoint(firstEndpoint, vertices);
+    S1Angle span = new S1Angle(firstEndpoint, secondEndpoint);
+    if (span.lessOrEquals(PROJECT_PERPENDICULAR_ERROR)) {
+      return new S2Point[] {firstEndpoint};
+    }
+
+    for (S2Point vertex : vertices) {
+      if (S2EdgeUtil.getDistance(vertex, firstEndpoint, secondEndpoint)
+          .greaterThan(PROJECT_PERPENDICULAR_ERROR)) {
+        return null;
+      }
+    }
+    return new S2Point[] {firstEndpoint, secondEndpoint};
+  }
+
+  private static S2Point farthestPoint(S2Point origin, List<S2Point> vertices) {
+    S2Point farthest = origin;
+    S1ChordAngle farthestDistance = S1ChordAngle.ZERO;
+    for (S2Point vertex : vertices) {
+      S1ChordAngle distance = new S1ChordAngle(origin, vertex);
+      if (distance.greaterThan(farthestDistance)) {
+        farthest = vertex;
+        farthestDistance = distance;
+      }
+    }
+    return farthest;
+  }
+
+  /**
+   * Writes a degenerate hull from its exact source coordinates. S2 still selects the spherical
+   * endpoint(s), but converting those points back to longitude/latitude would introduce
+   * floating-point drift into an otherwise unchanged input vertex.
+   */
+  private static Geography createDegenerateHull(
+      Geometry sourceGeometry, S2Point[] endpoints, int srid) {
+    SourceCoordinateIndex sourceIndex = new SourceCoordinateIndex(sourceGeometry.getCoordinates());
+    GeometryFactory factory = new GeometryFactory(new PrecisionModel(), srid);
+    Coordinate first = sourceIndex.resolve(endpoints[0]);
+    Geometry result;
+    if (endpoints.length == 1) {
+      result = factory.createPoint(first);
+    } else {
+      Coordinate second = sourceIndex.resolve(endpoints[1]);
+      result = factory.createLineString(new Coordinate[] {first, second});
+    }
+    return WKBGeography.fromJTS(result);
+  }
+
+  /**
+   * Writes a polygonal hull from the exact source coordinates selected by S2. For a non-degenerate
+   * hull, S2ConvexHullQuery returns a subset of its input vertices, so no computed vertex needs to
+   * be rounded back to longitude/latitude.
+   */
+  private static Geography createPolygonHull(Geometry sourceGeometry, S2Loop loop, int srid) {
+    SourceCoordinateIndex sourceIndex = new SourceCoordinateIndex(sourceGeometry.getCoordinates());
+    Coordinate[] hullCoordinates = new Coordinate[loop.numVertices() + 1];
+    for (int i = 0; i < loop.numVertices(); i++) {
+      hullCoordinates[i] = sourceIndex.resolve(loop.vertex(i));
+    }
+    hullCoordinates[loop.numVertices()] = new Coordinate(hullCoordinates[0]);
+
+    GeometryFactory factory = new GeometryFactory(new PrecisionModel(), srid);
+    return WKBGeography.fromJTS(factory.createPolygon(hullCoordinates));
+  }
+
+  /**
+   * Resolves the S2 vertices selected by the convex-hull query to their exact source coordinates.
+   * S2ConvexHullQuery retains input vertices for non-degenerate hulls, so the normal path is an
+   * exact hash lookup. The nearest-coordinate fallback covers directly constructed S2 Geography
+   * values whose JTS conversion may not reproduce the same S2Point bits.
+   */
+  private static final class SourceCoordinateIndex {
+    private final Coordinate[] sourceCoordinates;
+    private final Map<S2Point, Coordinate> exactCoordinates = new HashMap<>();
+
+    SourceCoordinateIndex(Coordinate[] sourceCoordinates) {
+      this.sourceCoordinates = sourceCoordinates;
+      for (Coordinate coordinate : sourceCoordinates) {
+        if (!Double.isFinite(coordinate.x) || !Double.isFinite(coordinate.y)) continue;
+        S2Point point = S2LatLng.fromDegrees(coordinate.y, coordinate.x).toPoint();
+        // Keep the first source spelling when equivalent coordinates map to the same S2 point.
+        exactCoordinates.putIfAbsent(point, new Coordinate(coordinate.x, coordinate.y));
+      }
+    }
+
+    Coordinate resolve(S2Point point) {
+      Coordinate coordinate = exactCoordinates.get(point);
+      if (coordinate != null) {
+        return new Coordinate(coordinate.x, coordinate.y);
+      }
+      return nearestSourceCoordinate(point, sourceCoordinates);
+    }
+  }
+
+  private static Coordinate nearestSourceCoordinate(
+      S2Point endpoint, Coordinate[] sourceCoordinates) {
+    Coordinate nearest = null;
+    double nearestDistance = Double.POSITIVE_INFINITY;
+    for (Coordinate coordinate : sourceCoordinates) {
+      if (!Double.isFinite(coordinate.x) || !Double.isFinite(coordinate.y)) continue;
+      S2Point candidate = S2LatLng.fromDegrees(coordinate.y, coordinate.x).toPoint();
+      double distance = endpoint.getDistance2(candidate);
+      if (distance < nearestDistance) {
+        nearest = coordinate;
+        nearestDistance = distance;
+      }
+    }
+    if (nearest == null) {
+      throw new IllegalArgumentException("Cannot construct a convex hull without finite vertices");
+    }
+    return new Coordinate(nearest.x, nearest.y);
+  }
+
   /**
    * Creates a line from two Point, MultiPoint, or LineString geographies. The returned geography
    * preserves the first input's SRID; its edges are interpreted as great-circle arcs by geography
diff --git a/common/src/test/java/org/apache/sedona/common/Geography/FunctionTest.java b/common/src/test/java/org/apache/sedona/common/Geography/FunctionTest.java
index c6019b7be1b..8f5431071ad 100644
--- a/common/src/test/java/org/apache/sedona/common/Geography/FunctionTest.java
+++ b/common/src/test/java/org/apache/sedona/common/Geography/FunctionTest.java
@@ -25,6 +25,7 @@
 import com.google.common.geometry.S2Loop;
 import com.google.common.geometry.S2Point;
 import org.apache.sedona.common.S2Geography.Geography;
+import org.apache.sedona.common.S2Geography.GeographyWKBSerializer;
 import org.apache.sedona.common.S2Geography.PolygonGeography;
 import org.apache.sedona.common.S2Geography.WKBGeography;
 import org.apache.sedona.common.geography.Constructors;
@@ -32,6 +33,8 @@
 import org.junit.Test;
 import org.locationtech.jts.geom.Coordinate;
 import org.locationtech.jts.geom.Geometry;
+import org.locationtech.jts.geom.LineString;
+import org.locationtech.jts.geom.MultiPoint;
 import org.locationtech.jts.geom.Point;
 import org.locationtech.jts.geom.Polygon;
 import org.locationtech.jts.geom.PrecisionModel;
@@ -221,6 +224,142 @@ public void xY_nonPointAndNull() throws ParseException {
     assertNull(Functions.y(null));
   }
 
+  @Test
+  public void convexHull_nullAndEmptyPreserveType() throws Exception {
+    assertNull(Functions.convexHull(null));
+
+    for (String wkt : new String[] {"POINT EMPTY", "LINESTRING EMPTY", "POLYGON EMPTY"}) {
+      Geography input = Constructors.geogFromWKT(wkt, 4326);
+      Geography hull = Functions.convexHull(input);
+      assertEquals(wkt, hull.toString());
+      assertEquals(4326, hull.getSRID());
+
+      Geography roundTripped =
+          GeographyWKBSerializer.deserialize(GeographyWKBSerializer.serialize(hull));
+      assertEquals(wkt, roundTripped.toString());
+      assertEquals(4326, roundTripped.getSRID());
+    }
+
+    WKBGeography legacyEmptyLine = WKBGeography.fromWKB(new byte[] {1, 2, 0, 0, 0}, 4326);
+    assertEquals(9, legacyEmptyLine.getWKBBytes().length);
+    Geography legacyHull = Functions.convexHull(legacyEmptyLine);
+    assertEquals("LINESTRING EMPTY", legacyHull.toString());
+    assertEquals(9, ((WKBGeography) legacyHull).getWKBBytes().length);
+  }
+
+  @Test
+  public void convexHull_degenerateInputsReturnPointOrLine() throws ParseException {
+    Geometry coincidentJts = new WKTReader().read("LINESTRING (1 2, 1 2, 1 2)");
+    coincidentJts.setSRID(4326);
+    Geography coincidentLine = WKBGeography.fromJTS(coincidentJts);
+    Geometry pointHull = Constructors.geogToGeometry(Functions.convexHull(coincidentLine));
+    assertTrue(pointHull instanceof Point);
+    assertEquals(1.0, ((Point) pointHull).getX(), 0.0);
+    assertEquals(2.0, ((Point) pointHull).getY(), 0.0);
+
+    Geography twoPoints = Constructors.geogFromWKT("MULTIPOINT ((0 0), (0 2))", 4326);
+    assertLineEndpoints(Functions.convexHull(twoPoints), 0, 0, 0, 2);
+
+    Geography preciseTwoPoints =
+        Constructors.geogFromWKT("MULTIPOINT ((-73.9857 40.7484), (-122.4194 37.7749))", 4326);
+    assertLineEndpoints(
+        Functions.convexHull(preciseTwoPoints), -73.9857, 40.7484, -122.4194, 37.7749, 0.0);
+
+    Geography collinear = Constructors.geogFromWKT("LINESTRING (0 0, 0 1, 0 2, 0 1)", 4326);
+    assertLineEndpoints(Functions.convexHull(collinear), 0, 0, 0, 2);
+
+    Geography fourCollinear =
+        Constructors.geogFromWKT("MULTIPOINT ((0 0), (10 0), (20 0), (30 0))", 4326);
+    assertLineEndpoints(Functions.convexHull(fourCollinear), 0, 0, 30, 0);
+
+    Geography equivalentLongitudes = Constructors.geogFromWKT("MULTIPOINT ((0 0), (360 0))", 4326);
+    assertTrue(
+        Constructors.geogToGeometry(Functions.convexHull(equivalentLongitudes)) instanceof Point);
+
+    Geography equivalentPoles =
+        Constructors.geogFromWKT("MULTIPOINT ((0 90), (120 90), (-45 90))", 4326);
+    assertTrue(Constructors.geogToGeometry(Functions.convexHull(equivalentPoles)) instanceof Point);
+
+    Geography collinearPolygon = Constructors.geogFromWKT("POLYGON ((0 0, 10 0, 20 0, 0 0))", 4326);
+    assertLineEndpoints(Functions.convexHull(collinearPolygon), 0, 0, 20, 0);
+
+    Geography collectionWithCollinearPolygon =
+        Constructors.geogFromWKT(
+            "GEOMETRYCOLLECTION (POLYGON ((0 0, 10 0, 20 0, 0 0)), POINT (30 0))", 4326);
+    assertLineEndpoints(Functions.convexHull(collectionWithCollinearPolygon), 0, 0, 30, 0);
+  }
+
+  @Test
+  public void convexHull_polygonSkipsHolesAndPreservesSrid() throws ParseException {
+    Geography input =
+        Constructors.geogFromWKT(
+            "POLYGON ((0 0, 2 0, 0 2, 0 0), (0.2 0.2, 0.4 0.2, 0.2 0.4, 0.2 0.2))", 4326);
+    Geography hull = Functions.convexHull(input);
+    Geometry jts = Constructors.geogToGeometry(hull);
+
+    assertTrue(jts instanceof Polygon);
+    assertEquals(0, ((Polygon) jts).getNumInteriorRing());
+    assertEquals(4, ((Polygon) jts).getExteriorRing().getNumPoints());
+    assertEquals(4326, hull.getSRID());
+  }
+
+  @Test
+  public void convexHull_polygonPreservesExactSourceCoordinates() throws ParseException {
+    Geography input =
+        Constructors.geogFromWKT("MULTIPOINT ((0 0), (10 0), (10 10), (0 10), (5 5))", 4326);
+    Geometry hull = Constructors.geogToGeometry(Functions.convexHull(input));
+
+    assertTrue(hull instanceof Polygon);
+    assertEquals(5, hull.getNumPoints());
+    for (Coordinate coordinate : hull.getCoordinates()) {
+      boolean isExactSourceVertex =
+          (coordinate.x == 0.0 && coordinate.y == 0.0)
+              || (coordinate.x == 10.0 && coordinate.y == 0.0)
+              || (coordinate.x == 10.0 && coordinate.y == 10.0)
+              || (coordinate.x == 0.0 && coordinate.y == 10.0);
+      assertTrue(
+          "Hull vertex drifted from its source coordinate: " + coordinate, isExactSourceVertex);
+    }
+  }
+
+  @Test
+  public void convexHull_collectionAndAntimeridianAreSpherical() throws ParseException {
+    Geography collection =
+        Constructors.geogFromWKT(
+            "GEOMETRYCOLLECTION (POINT (0 0), LINESTRING (1 0, 5 5), "
+                + "POLYGON ((0 1, 0.2 0.2, 1 0, 0 1)))",
+            4326);
+    assertTrue(Constructors.geogToGeometry(Functions.convexHull(collection)) instanceof Polygon);
+
+    Geography acrossDateLine =
+        Constructors.geogFromWKT("MULTIPOINT ((170 -10), (170 10), (-170 10), (-170 -10))", 4326);
+    Geography hull = Functions.convexHull(acrossDateLine);
+    assertTrue(Constructors.geogToGeometry(hull) instanceof Polygon);
+    assertTrue("expected the small antimeridian hull", Functions.area(hull) < 1e14);
+  }
+
+  @Test
+  public void convexHull_fullSphereIsRejected() throws ParseException {
+    Geography input = Constructors.geogFromWKT("MULTIPOINT ((0 0), (120 0), (-120 0))", 4326);
+    try {
+      Functions.convexHull(input);
+      fail("Expected a full-sphere hull to be rejected");
+    } catch (UnsupportedOperationException e) {
+      assertTrue(e.getMessage().contains("full sphere"));
+    }
+  }
+
+  @Test
+  public void convexHull_publicPolygonUsesNormalizedShellInterior() throws ParseException {
+    Geography polygon = Constructors.geogFromWKT("POLYGON ((0 0, 0 10, 10 0, 0 0))", 4326);
+    Geography hull = Functions.convexHull(polygon);
+
+    assertEquals("ST_Polygon", Functions.geometryType(hull));
+    assertTrue(Functions.area(hull) < 1.0e13);
+    assertTrue(Functions.contains(hull, Constructors.geogFromWKT("POINT (1 1)", 4326)));
+    assertFalse(Functions.contains(hull, Constructors.geogFromWKT("POINT (20 20)", 4326)));
+  }
+
   @Test
   public void asEWKT_usesStoredWKBForOrdinaryGeography() throws ParseException {
     Geography geography = Constructors.geogFromWKT("POINT (-122.4194 37.7749)", 4326);
@@ -344,6 +483,52 @@ public void makeLine_multiPointAndLineString() throws ParseException {
     }
   }
 
+  @Test
+  public void createMultiGeography_preservesMembersAndFirstSrid() throws ParseException {
+    Geography first = Constructors.geogFromWKT("POINT (1 2)", 4326);
+    Geography duplicate = Constructors.geogFromWKT("POINT (1 2)", 3857);
+    Geography second = Constructors.geogFromWKT("POINT (3 4)", 4326);
+    Geography collected =
+        Functions.createMultiGeography(new Geography[] {first, null, duplicate, second});
+    Geometry jts = Constructors.geogToGeometry(collected);
+
+    assertTrue(jts instanceof MultiPoint);
+    assertEquals(3, jts.getNumGeometries());
+    assertEquals(4326, collected.getSRID());
+
+    Geography line = Constructors.geogFromWKT("LINESTRING (0 0, 1 1)", 4326);
+    Geography mixed = Functions.createMultiGeography(new Geography[] {first, line});
+    assertEquals("GeometryCollection", Constructors.geogToGeometry(mixed).getGeometryType());
+
+    Geography empty = Functions.createMultiGeography(new Geography[] {null});
+    assertEquals("GEOMETRYCOLLECTION EMPTY", Functions.asText(empty));
+  }
+
+  private static void assertLineEndpoints(
+      Geography geography, double x0, double y0, double x1, double y1) {
+    assertLineEndpoints(geography, x0, y0, x1, y1, EPS);
+  }
+
+  private static void assertLineEndpoints(
+      Geography geography, double x0, double y0, double x1, double y1, double tolerance) {
+    Geometry geometry = Constructors.geogToGeometry(geography);
+    assertTrue(geometry instanceof LineString);
+    LineString line = (LineString) geometry;
+    Coordinate start = line.getCoordinateN(0);
+    Coordinate end = line.getCoordinateN(line.getNumPoints() - 1);
+    boolean forward =
+        Math.abs(start.x - x0) <= tolerance
+            && Math.abs(start.y - y0) <= tolerance
+            && Math.abs(end.x - x1) <= tolerance
+            && Math.abs(end.y - y1) <= tolerance;
+    boolean reverse =
+        Math.abs(start.x - x1) <= tolerance
+            && Math.abs(start.y - y1) <= tolerance
+            && Math.abs(end.x - x0) <= tolerance
+            && Math.abs(end.y - y0) <= tolerance;
+    assertTrue("Unexpected line endpoints: " + geometry, forward || reverse);
+  }
+
   @Test
   public void makeLine_antimeridianUsesGreatCircleLength() throws ParseException {
     Geography start = Constructors.geogFromWKT("POINT (179 0)", 4326);
diff --git a/docs/api/flink/Aggregate-Functions/ST_Collect_Agg.md b/docs/api/flink/Aggregate-Functions/ST_Collect_Agg.md
new file mode 100644
index 00000000000..a473142f4d0
--- /dev/null
+++ b/docs/api/flink/Aggregate-Functions/ST_Collect_Agg.md
@@ -0,0 +1,67 @@
+<!--
+ 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.
+ -->
+
+# ST_Collect_Agg
+
+Introduction: Collects all non-null spatial values in a group without dissolving their
+boundaries. Homogeneous `Point`, `LineString`, or `Polygon` values produce the corresponding
+multi-object; mixed types produce a `GeometryCollection`.
+
+![ST_Collect_Agg](../../../image/ST_Collect_Agg/ST_Collect_Agg.svg "ST_Collect_Agg")
+
+Format:
+
+`ST_Collect_Agg (A: Geometry)`
+
+`ST_Collect_Agg (A: Geography)`
+
+Return type: `Geometry` or `Geography`, matching the input type
+
+Since: `v1.9.1`
+
+Null values are ignored and duplicates are preserved. The function returns `NULL` when a group
+has no non-null values. All `Geography` values in a group must have the same SRID.
+
+`ST_Collect_Aggr` is also registered as an alias.
+
+Geometry example:
+
+```sql
+SELECT ST_Collect_Agg(geom)
+FROM (
+    VALUES
+        (ST_GeomFromWKT('POINT(1 2)')),
+        (ST_GeomFromWKT('POINT(3 4)')),
+        (ST_GeomFromWKT('POINT(1 2)'))
+) AS observations(geom)
+```
+
+Output:
+
+```
+MULTIPOINT ((1 2), (3 4), (1 2))
+```
+
+Geography example:
+
+```sql
+SELECT region, ST_Collect_Agg(geog) AS geographies
+FROM observations
+GROUP BY region
+```
diff --git a/docs/api/flink/Geography-Functions.md b/docs/api/flink/Geography-Functions.md
index 0f9b772714e..55a648f1513 100644
--- a/docs/api/flink/Geography-Functions.md
+++ b/docs/api/flink/Geography-Functions.md
@@ -41,7 +41,7 @@ These functions create geography objects from various formats, or convert betwee
 
 ## Geography Functions
 
-These functions edit, measure, or format geography objects. Measurements are computed on a spherical model of the Earth (radius `R = 6 371 008 m`, the mean Earth radius), not the WGS84 ellipsoid — areas in square meters and lengths/distances in meters. (`ST_Buffer` is the exception: it is computed on the WGS84 spheroid.)
+These functions process, collect, measure, or format geography objects. Measurements are computed on a spherical model of the Earth (radius `R = 6 371 008 m`, the mean Earth radius), not the WGS84 ellipsoid — areas in square meters and lengths/distances in meters. (`ST_Buffer` is the exception: it is computed on the WGS84 spheroid.)
 
 | Function | Return type | Description | Since |
 | :--- | :--- | :--- | :--- |
@@ -50,6 +50,8 @@ These functions edit, measure, or format geography objects. Measurements are com
 | [ST_AsText](Geography-Functions/ST_AsText.md) | String | Return the WKT representation of a geography. | v1.9.1 |
 | [ST_Buffer](Geography-Functions/ST_Buffer.md) | Geography | Return the geodesic buffer of a geography (distance in meters). | v1.9.1 |
 | [ST_Centroid](Geography-Functions/ST_Centroid.md) | Geography | Return the centroid of a geography as a Geography point. | v1.9.1 |
+| [ST_Collect](Geometry-Editors/ST_Collect.md) | Geography | Collect two geographies or an array of geographies into a multi-object or collection without dissolving boundaries. | v1.9.1 |
+| [ST_ConvexHull](Geometry-Processing/ST_ConvexHull.md) | Geography | Return the spherical convex hull of a geography with geodesic edges. | v1.9.1 |
 | [ST_Distance](Geography-Functions/ST_Distance.md) | Double | Return the minimum geodesic distance between two geographies in meters. | v1.9.1 |
 | [ST_Envelope](Geography-Functions/ST_Envelope.md) | Geography | Return the bounding box of a geography. Supports antimeridian splitting. | v1.9.1 |
 | [ST_GeometryType](Geography-Functions/ST_GeometryType.md) | String | Return the type of a geography as a string. | v1.9.1 |
@@ -60,6 +62,14 @@ These functions edit, measure, or format geography objects. Measurements are com
 | [ST_X](Geography-Functions/ST_X.md) | Double | Return the longitude (X coordinate) of a point geography. | v1.9.1 |
 | [ST_Y](Geography-Functions/ST_Y.md) | Double | Return the latitude (Y coordinate) of a point geography. | v1.9.1 |
 
+## Geography Aggregate Functions
+
+These functions aggregate groups of geography values.
+
+| Function | Return type | Description | Since |
+| :--- | :--- | :--- | :--- |
+| [ST_Collect_Agg](Aggregate-Functions/ST_Collect_Agg.md) | Geography | Collect all non-null geographies in a group without dissolving boundaries. All inputs must have the same SRID. | v1.9.1 |
+
 ## Geography Predicates
 
 These functions test spatial relationships between two geographies on the sphere.
diff --git a/docs/api/flink/Geometry-Editors/ST_Collect.md b/docs/api/flink/Geometry-Editors/ST_Collect.md
index 5d15e1edc1f..ed08fb419de 100644
--- a/docs/api/flink/Geometry-Editors/ST_Collect.md
+++ b/docs/api/flink/Geometry-Editors/ST_Collect.md
@@ -19,21 +19,30 @@
 
 # ST_Collect
 
-Introduction: Returns MultiGeometry object based on geometry column/s or array with geometries
+Introduction: Collects spatial values without dissolving their boundaries. Homogeneous
+`Point`, `LineString`, or `Polygon` values produce the corresponding multi-object; mixed types
+produce a `GeometryCollection`.
 
 ![ST_Collect](../../../image/ST_Collect/ST_Collect.svg "ST_Collect")
 
 Format:
 
-`ST_Collect(*geom: Geometry)`
+`ST_Collect(geom1: Geometry, geom2: Geometry)`
 
 `ST_Collect(geom: ARRAY[Geometry])`
 
-Return type: `Geometry`
+`ST_Collect(geog1: Geography, geog2: Geography)`
 
-Since: `v1.5.0`
+`ST_Collect(geog: ARRAY[Geography])`
 
-Example:
+Return type: `Geometry` or `Geography`, matching the input type
+
+Since: `v1.5.0` (`Geometry`), `v1.9.1` (`Geography`)
+
+Member order and duplicates are retained. The `Geography` overloads ignore null elements and use
+the first non-null input's SRID for the result.
+
+Geometry example:
 
 ```sql
 SELECT ST_Collect(
@@ -52,23 +61,21 @@ Result:
 +---------------------------------------------------------------+
 ```
 
-Example:
+Geography array example:
 
 ```sql
-SELECT ST_Collect(
-    Array(
-        ST_GeomFromText('POINT(21.427834 52.042576573)'),
-        ST_GeomFromText('POINT(45.342524 56.342354355)')
+SELECT ST_AsEWKT(
+    ST_Collect(
+        ARRAY[
+            ST_GeogFromWKT('POINT(-122.4 37.8)', 4326),
+            ST_GeogFromWKT('POINT(-122.5 37.7)', 4326)
+        ]
     )
-) AS geom
+) AS geog
 ```
 
 Result:
 
 ```
-+---------------------------------------------------------------+
-|geom                                                           |
-+---------------------------------------------------------------+
-|MULTIPOINT ((21.427834 52.042576573), (45.342524 56.342354355))|
-+---------------------------------------------------------------+
+SRID=4326;MULTIPOINT ((-122.4 37.8), (-122.5 37.7))
 ```
diff --git a/docs/api/flink/Geometry-Functions.md b/docs/api/flink/Geometry-Functions.md
index f3e9d279f90..1dee6619226 100644
--- a/docs/api/flink/Geometry-Functions.md
+++ b/docs/api/flink/Geometry-Functions.md
@@ -109,7 +109,7 @@ These functions create modified geometries by changing type, structure, or verti
 | Function | Return type | Description | Since |
 | :--- | :--- | :--- | :--- |
 | [ST_AddPoint](Geometry-Editors/ST_AddPoint.md) | Geometry | Return Linestring with additional point at the given index, if position is not available the point will be added at the end of line. | v1.3.0 |
-| [ST_Collect](Geometry-Editors/ST_Collect.md) | Geometry | Returns MultiGeometry object based on geometry column/s or array with geometries | v1.5.0 |
+| [ST_Collect](Geometry-Editors/ST_Collect.md) | Geometry or Geography | Collects two spatial values or an array of spatial values into a multi-object or collection without dissolving boundaries. | v1.5.0 |
 | [ST_CollectionExtract](Geometry-Editors/ST_CollectionExtract.md) | Geometry | Returns a homogeneous multi-geometry from a given geometry collection. | v1.5.0 |
 | [ST_FlipCoordinates](Geometry-Editors/ST_FlipCoordinates.md) | Geometry | Returns a version of the given geometry with X and Y axis flipped. | v1.2.0 |
 | [ST_Force2D](Geometry-Editors/ST_Force2D.md) | Geometry | Forces the geometries into a "2-dimensional mode" so that all output representations will only have the X and Y coordinates. This function is an alias of [ST_Force_2D](Geometry-Editors/ST_Force_2D.md). | v1.8.0 |
@@ -216,7 +216,7 @@ These functions compute geometric constructions, or alter geometry size or shape
 | [ST_BuildArea](Geometry-Processing/ST_BuildArea.md) | Geometry | Returns the areal geometry formed by the constituent linework of the input geometry. | v1.2.1 |
 | [ST_Centroid](Geometry-Processing/ST_Centroid.md) | Geometry | Return the centroid point of A | v1.5.0 |
 | [ST_ConcaveHull](Geometry-Processing/ST_ConcaveHull.md) | Geometry | Return the Concave Hull of polygon A, with alpha set to pctConvex[0, 1] in the Delaunay Triangulation method, the concave hull will not contain a hole unless allowHoles is set to true | v1.4.0 |
-| [ST_ConvexHull](Geometry-Processing/ST_ConvexHull.md) | Geometry | Return the Convex Hull of polygon A | v1.5.0 |
+| [ST_ConvexHull](Geometry-Processing/ST_ConvexHull.md) | Geometry or Geography | Returns the planar convex hull of a Geometry or spherical convex hull of a Geography. | v1.5.0 |
 | [ST_DelaunayTriangles](Geometry-Processing/ST_DelaunayTriangles.md) | Geometry | This function computes the [Delaunay triangulation](https://en.wikipedia.org/wiki/Delaunay_triangulation) for the set of vertices in the input geometry. An optional `tolerance` parameter allows sna... | v1.6.1 |
 | [ST_GeneratePoints](Geometry-Processing/ST_GeneratePoints.md) | Geometry | Generates a specified quantity of pseudo-random points within the boundaries of the provided polygonal geometry. When `seed` is either zero or not defined then output will be random. | v1.6.1 |
 | [ST_GeometricMedian](Geometry-Processing/ST_GeometricMedian.md) | Geometry | Computes the approximate geometric median of a MultiPoint geometry using the Weiszfeld algorithm. The geometric median provides a centrality measure that is less sensitive to outlier points than th... | v1.4.1 |
@@ -266,10 +266,11 @@ These functions change the position and shape of geometries using affine transfo
 
 ## Aggregate Functions
 
-These functions perform aggregate operations on groups of geometries.
+These functions perform aggregate operations on groups of spatial values.
 
 | Function | Return type | Description | Since |
 | :--- | :--- | :--- | :--- |
+| [ST_Collect_Agg](Aggregate-Functions/ST_Collect_Agg.md) | Geometry or Geography | Collects all non-null spatial values in a group without dissolving boundaries. Geography inputs must share an SRID. | v1.9.1 |
 | [ST_Envelope_Agg](Aggregate-Functions/ST_Envelope_Agg.md) | Geometry | Return the entire envelope boundary of all geometries in A. Empty geometries and null values are skipped. If all inputs are empty or null, the result is null. This behavior is consistent with PostG... | v1.3.0 |
 | [ST_Intersection_Agg](Aggregate-Functions/ST_Intersection_Agg.md) | Geometry | Return the polygon intersection of all polygons in A | v1.5.0 |
 | [ST_Union_Agg](Aggregate-Functions/ST_Union_Agg.md) | Geometry | Return the polygon union of all polygons in A. All inputs must be polygons. | v1.3.0 |
diff --git a/docs/api/flink/Geometry-Processing/ST_ConvexHull.md b/docs/api/flink/Geometry-Processing/ST_ConvexHull.md
index 617ff90d4c7..14233ed9e33 100644
--- a/docs/api/flink/Geometry-Processing/ST_ConvexHull.md
+++ b/docs/api/flink/Geometry-Processing/ST_ConvexHull.md
@@ -19,17 +19,27 @@
 
 # ST_ConvexHull
 
-Introduction: Return the Convex Hull of polygon A
+Introduction: Returns the smallest convex region containing the input. `Geometry` hulls are
+computed in the coordinate plane. `Geography` hulls are computed on the sphere, with geodesic
+edges.
 
 ![ST_ConvexHull](../../../image/ST_ConvexHull/ST_ConvexHull.svg "ST_ConvexHull")
 
-Format: `ST_ConvexHull (A: Geometry)`
+Format:
 
-Return type: `Geometry`
+`ST_ConvexHull (A: Geometry)`
 
-Since: `v1.5.0`
+`ST_ConvexHull (A: Geography)`
 
-Example:
+Return type: `Geometry` or `Geography`, matching the input type
+
+Since: `v1.5.0` (`Geometry`), `v1.9.1` (`Geography`)
+
+For `Geography`, one unique point produces a `Point`, collinear points produce a `LineString`,
+and all other non-empty inputs produce a spherical `Polygon`. Empty inputs retain their type and
+SRID. A hull covering the full sphere is not supported.
+
+Geometry example:
 
 ```sql
 SELECT ST_ConvexHull(ST_GeomFromText('POLYGON((175 150, 20 40, 50 60, 125 100, 175 150))'))
@@ -40,3 +50,22 @@ Output:
 ```
 POLYGON ((20 40, 175 150, 125 100, 20 40))
 ```
+
+Geography example:
+
+```sql
+SELECT ST_GeometryType(
+    ST_ConvexHull(
+        ST_GeogFromWKT(
+            'MULTIPOINT ((170 -10), (170 10), (-170 10), (-170 -10))',
+            4326
+        )
+    )
+)
+```
+
+Output:
+
+```
+ST_Polygon
+```
diff --git a/docs/api/sql/Aggregate-Functions/ST_Collect_Agg.md b/docs/api/sql/Aggregate-Functions/ST_Collect_Agg.md
index 5f8081f75f1..76415b03d6a 100644
--- a/docs/api/sql/Aggregate-Functions/ST_Collect_Agg.md
+++ b/docs/api/sql/Aggregate-Functions/ST_Collect_Agg.md
@@ -19,15 +19,27 @@
 
 # ST_Collect_Agg
 
-Introduction: Collects all geometries in a geometry column into a single multi-geometry (MultiPoint, MultiLineString, MultiPolygon, or GeometryCollection). Unlike `ST_Union_Agg`, this function does not dissolve boundaries between geometries - it simply collects them into a multi-geometry.
+Introduction: Collects all non-null spatial values in a column into a single multi-object or
+collection. Unlike `ST_Union_Agg`, this function does not dissolve boundaries. Homogeneous
+`Point`, `LineString`, or `Polygon` inputs produce the corresponding multi-object; mixed input
+types produce the OGC/WKT `GeometryCollection` type.
 
 ![ST_Collect_Agg](../../../image/ST_Collect_Agg/ST_Collect_Agg.svg "ST_Collect_Agg")
 
-Format: `ST_Collect_Agg (A: geometryColumn)`
+Format:
 
-Return type: `Geometry`
+`ST_Collect_Agg (A: Geometry)`
 
-Since: `v1.8.1`
+`ST_Collect_Agg (A: Geography)`
+
+Return type: `Geometry` or `Geography`, matching the input type
+
+Since: `v1.8.1` (`Geometry`), `v1.9.1` (`Geography`)
+
+Duplicates are preserved. The function returns `NULL` when a group has no non-null values.
+All non-null `Geography` values in a group must have the same SRID; a group containing mixed
+SRIDs is rejected. In contrast, scalar [`ST_Collect`](../Geometry-Editors/ST_Collect.md) uses the
+first non-null Geography input's SRID for the output.
 
 SQL Example
 
@@ -52,3 +64,18 @@ SQL Example with GROUP BY
 ```sql
 SELECT category, ST_Collect_Agg(geom) FROM geometries GROUP BY category
 ```
+
+Geography SQL Example with GROUP BY
+
+```sql
+WITH observations AS (
+    SELECT 'west' AS region, ST_GeogFromWKT('POINT(-122.4 37.8)', 4326) AS geog
+    UNION ALL
+    SELECT 'west' AS region, ST_GeogFromWKT('POINT(-122.5 37.7)', 4326) AS geog
+    UNION ALL
+    SELECT 'east' AS region, ST_GeogFromWKT('POINT(-73.9 40.7)', 4326) AS geog
+)
+SELECT region, ST_Collect_Agg(geog) AS geographies
+FROM observations
+GROUP BY region
+```
diff --git a/docs/api/sql/Geometry-Editors/ST_Collect.md b/docs/api/sql/Geometry-Editors/ST_Collect.md
index 768551fe6a0..31f4b76271a 100644
--- a/docs/api/sql/Geometry-Editors/ST_Collect.md
+++ b/docs/api/sql/Geometry-Editors/ST_Collect.md
@@ -19,7 +19,9 @@
 
 # ST_Collect
 
-Introduction: Returns MultiGeometry object based on geometry column/s or array with geometries
+Introduction: Collects spatial values into a multi-object or collection without dissolving their
+boundaries. Homogeneous `Point`, `LineString`, or `Polygon` inputs produce the corresponding
+multi-object; mixed input types produce the OGC/WKT `GeometryCollection` type.
 
 ![ST_Collect](../../../image/ST_Collect/ST_Collect.svg "ST_Collect")
 
@@ -29,9 +31,18 @@ Format:
 
 `ST_Collect(geom: ARRAY[Geometry])`
 
-Return type: `Geometry`
+`ST_Collect(*geog: Geography)`
 
-Since: `v1.2.0`
+`ST_Collect(geog: ARRAY[Geography])`
+
+Return type: `Geometry` or `Geography`, matching the input type
+
+Since: `v1.2.0` (`Geometry`), `v1.9.1` (`Geography`)
+
+Null elements are ignored. Member values and duplicates are preserved. For `Geography` inputs,
+the first non-null input supplies the output SRID; later inputs are not required to have the same
+SRID. In contrast, [`ST_Collect_Agg`](../Aggregate-Functions/ST_Collect_Agg.md) rejects a group
+containing mixed Geography SRIDs.
 
 SQL Example
 
@@ -52,6 +63,21 @@ Result:
 +---------------------------------------------------------------+
 ```
 
+Geography SQL Example with GROUP BY
+
+```sql
+WITH observations AS (
+    SELECT 'west' AS region, ST_GeogFromWKT('POINT(-122.4 37.8)', 4326) AS geog
+    UNION ALL
+    SELECT 'west' AS region, ST_GeogFromWKT('POINT(-122.5 37.7)', 4326) AS geog
+    UNION ALL
+    SELECT 'east' AS region, ST_GeogFromWKT('POINT(-73.9 40.7)', 4326) AS geog
+)
+SELECT region, ST_Collect(ARRAY_AGG(geog)) AS geographies
+FROM observations
+GROUP BY region
+```
+
 SQL Example
 
 ```sql
diff --git a/docs/api/sql/Geometry-Processing/ST_ConvexHull.md b/docs/api/sql/Geometry-Processing/ST_ConvexHull.md
index 8e67a57670e..2292eebfada 100644
--- a/docs/api/sql/Geometry-Processing/ST_ConvexHull.md
+++ b/docs/api/sql/Geometry-Processing/ST_ConvexHull.md
@@ -19,15 +19,26 @@
 
 # ST_ConvexHull
 
-Introduction: Return the Convex Hull of polygon A
+Introduction: Returns the smallest convex region containing A. For `Geometry` inputs, the hull is
+computed in the coordinate plane. For `Geography` inputs, it is computed on the sphere and its
+edges follow geodesics.
 
 ![ST_ConvexHull](../../../image/ST_ConvexHull/ST_ConvexHull.svg "ST_ConvexHull")
 
-Format: `ST_ConvexHull (A: Geometry)`
+Format:
 
-Return type: `Geometry`
+`ST_ConvexHull (A: Geometry)`
 
-Since: `v1.0.0`
+`ST_ConvexHull (A: Geography)`
+
+Return type: `Geometry` or `Geography`, matching the input type
+
+Since: `v1.0.0` (`Geometry`), `v1.9.1` (`Geography`)
+
+For a `Geography` input, a single unique point produces a `Point`, while collinear points produce
+a `LineString`. Empty inputs preserve their input type. Polygon holes do not change the hull.
+If the spherical convex hull is the full sphere, the function throws an
+`UnsupportedOperationException` because the full sphere cannot be represented by OGC WKB.
 
 SQL Example
 
@@ -40,3 +51,22 @@ Output:
 ```
 POLYGON ((20 40, 175 150, 125 100, 20 40))
 ```
+
+Geography SQL Example
+
+```sql
+SELECT ST_GeometryType(
+    ST_ConvexHull(
+        ST_GeogFromWKT(
+            'MULTIPOINT ((170 -10), (170 10), (-170 10), (-170 -10))',
+            4326
+        )
+    )
+)
+```
+
+Output:
+
+```
+ST_Polygon
+```
diff --git a/docs/api/sql/geography/Geography-Functions.md b/docs/api/sql/geography/Geography-Functions.md
index 92755121126..950ebfbd443 100644
--- a/docs/api/sql/geography/Geography-Functions.md
+++ b/docs/api/sql/geography/Geography-Functions.md
@@ -46,6 +46,9 @@ These functions operate on geography type objects.
 | [ST_AsText](Geography-Functions/ST_AsText.md) | String | Return the Well-Known Text (WKT) representation of a geography. | v1.9.1 |
 | [ST_Centroid](Geography-Functions/ST_Centroid.md) | Geography | Return the planar centroid of a geography as a Geography point (computed in projected lon/lat space). | v1.9.1 |
 | [ST_Buffer](Geography-Functions/ST_Buffer.md) | Geography | Return the metric ε-buffer of a geography. Distance is always interpreted as meters along the spheroid. | v1.9.1 |
+| [ST_Collect](../Geometry-Editors/ST_Collect.md) | Geography | Collect Geography values without dissolving boundaries, returning a `MultiPoint`, `MultiLineString`, `MultiPolygon`, or OGC/WKT `GeometryCollection`. The first non-null input supplies the output SRID. | v1.9.1 |
+| [ST_Collect_Agg](../Aggregate-Functions/ST_Collect_Agg.md) | Geography | Aggregate Geography values without dissolving boundaries. All non-null values in a group must have the same SRID. | v1.9.1 |
+| [ST_ConvexHull](../Geometry-Processing/ST_ConvexHull.md) | Geography | Return the spherical convex hull with geodesic edges. Full-sphere hulls are not supported because they cannot be represented by OGC WKB. | v1.9.1 |
 | [ST_Envelope](Geography-Functions/ST_Envelope.md) | Geography | Return the bounding box (envelope) of a geography. Supports anti-meridian splitting. | v1.8.0 |
 | [ST_GeometryType](Geography-Functions/ST_GeometryType.md) | String | Return the type of a geography as a string (e.g., "ST_Point", "ST_Polygon"). | v1.9.1 |
 | [ST_MakeLine](Geography-Functions/ST_MakeLine.md) | Geography | Create a geography LineString from two Point, MultiPoint, or LineString geographies. | v1.9.1 |
diff --git a/flink/src/main/java/org/apache/sedona/flink/Catalog.java b/flink/src/main/java/org/apache/sedona/flink/Catalog.java
index ca5ed46ea61..49391445c23 100644
--- a/flink/src/main/java/org/apache/sedona/flink/Catalog.java
+++ b/flink/src/main/java/org/apache/sedona/flink/Catalog.java
@@ -30,10 +30,12 @@ public static UserDefinedFunction[] getFuncs() {
       new Aggregators.ST_3DExtent(),
       new Aggregators.ST_Intersection_Aggr(),
       new Aggregators.ST_Union_Aggr(),
+      new Aggregators.ST_Collect_Aggr(),
       // Aliases for *_Aggr functions with *_Agg suffix
       new Aggregators.ST_Envelope_Agg(),
       new Aggregators.ST_Intersection_Agg(),
       new Aggregators.ST_Union_Agg(),
+      new Aggregators.ST_Collect_Agg(),
       new Constructors.ST_Point(),
       new Constructors.ST_PointZ(),
       new Constructors.ST_PointM(),
diff --git a/flink/src/main/java/org/apache/sedona/flink/expressions/Accumulators.java b/flink/src/main/java/org/apache/sedona/flink/expressions/Accumulators.java
index 9550934db92..68079adb5be 100644
--- a/flink/src/main/java/org/apache/sedona/flink/expressions/Accumulators.java
+++ b/flink/src/main/java/org/apache/sedona/flink/expressions/Accumulators.java
@@ -18,6 +18,9 @@
  */
 package org.apache.sedona.flink.expressions;
 
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
 import org.apache.flink.table.annotation.DataTypeHint;
 import org.apache.sedona.flink.GeometryTypeSerializer;
 import org.locationtech.jts.geom.Geometry;
@@ -79,4 +82,24 @@ public static class AccGeometryN {
       this.numGeoms = numGeoms;
     }
   }
+
+  /**
+   * RAW accumulator used by ST_Collect_Aggr for both Geometry and Geography input.
+   *
+   * <p>Values are stored in their stable binary forms so that the accumulator does not depend on
+   * either spatial object's internal representation.
+   */
+  public static class AccGeometryCollection implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    public List<byte[]> values = new ArrayList<>();
+    public Boolean geography;
+    public int srid;
+
+    void reset() {
+      values.clear();
+      geography = null;
+      srid = 0;
+    }
+  }
 }
diff --git a/flink/src/main/java/org/apache/sedona/flink/expressions/Aggregators.java b/flink/src/main/java/org/apache/sedona/flink/expressions/Aggregators.java
index ff13fd8e739..14c55a0c93f 100644
--- a/flink/src/main/java/org/apache/sedona/flink/expressions/Aggregators.java
+++ b/flink/src/main/java/org/apache/sedona/flink/expressions/Aggregators.java
@@ -18,12 +18,18 @@
  */
 package org.apache.sedona.flink.expressions;
 
+import java.io.IOException;
+import java.util.Arrays;
 import org.apache.flink.table.annotation.DataTypeHint;
+import org.apache.flink.table.annotation.FunctionHint;
 import org.apache.flink.table.functions.AggregateFunction;
+import org.apache.sedona.common.S2Geography.Geography;
+import org.apache.sedona.common.S2Geography.GeographyWKBSerializer;
 import org.apache.sedona.common.geometryObjects.Box2D;
 import org.apache.sedona.common.geometryObjects.Box3D;
 import org.apache.sedona.flink.Box2DTypeSerializer;
 import org.apache.sedona.flink.Box3DTypeSerializer;
+import org.apache.sedona.flink.GeographyTypeSerializer;
 import org.apache.sedona.flink.GeometryTypeSerializer;
 import org.locationtech.jts.geom.Coordinate;
 import org.locationtech.jts.geom.Envelope;
@@ -372,6 +378,167 @@ public void resetAccumulator(Accumulators.AccGeometry acc) {
     }
   }
 
+  /**
+   * Collects all non-null Geometry or Geography inputs while preserving order and duplicates.
+   *
+   * <p>The output spatial type follows the input type. Geography values in a group must use the
+   * same SRID.
+   */
+  @FunctionHint(
+      input =
+          @DataTypeHint(
+              value = "RAW",
+              rawSerializer = GeometryTypeSerializer.class,
+              bridgedTo = Geometry.class),
+      accumulator =
+          @DataTypeHint(value = "RAW", bridgedTo = Accumulators.AccGeometryCollection.class),
+      output =
+          @DataTypeHint(
+              value = "RAW",
+              rawSerializer = GeometryTypeSerializer.class,
+              bridgedTo = Geometry.class))
+  @FunctionHint(
+      input =
+          @DataTypeHint(
+              value = "RAW",
+              rawSerializer = GeographyTypeSerializer.class,
+              bridgedTo = Geography.class),
+      accumulator =
+          @DataTypeHint(value = "RAW", bridgedTo = Accumulators.AccGeometryCollection.class),
+      output =
+          @DataTypeHint(
+              value = "RAW",
+              rawSerializer = GeographyTypeSerializer.class,
+              bridgedTo = Geography.class))
+  public static class ST_Collect_Aggr
+      extends AggregateFunction<Object, Accumulators.AccGeometryCollection> {
+
+    @Override
+    public Accumulators.AccGeometryCollection createAccumulator() {
+      return new Accumulators.AccGeometryCollection();
+    }
+
+    @Override
+    public Object getValue(Accumulators.AccGeometryCollection acc) {
+      if (acc.values.isEmpty()) {
+        return null;
+      }
+      if (Boolean.TRUE.equals(acc.geography)) {
+        Geography[] geographies = new Geography[acc.values.size()];
+        try {
+          for (int i = 0; i < geographies.length; i++) {
+            geographies[i] = GeographyWKBSerializer.deserialize(acc.values.get(i));
+          }
+        } catch (IOException e) {
+          throw new IllegalStateException("Failed to deserialize Geography collection", e);
+        }
+        return org.apache.sedona.common.geography.Functions.createMultiGeography(geographies);
+      }
+
+      Geometry[] geometries = new Geometry[acc.values.size()];
+      for (int i = 0; i < geometries.length; i++) {
+        geometries[i] =
+            org.apache.sedona.common.geometrySerde.GeometrySerializer.deserialize(
+                acc.values.get(i));
+      }
+      return org.apache.sedona.common.Functions.createMultiGeometry(geometries);
+    }
+
+    public void accumulate(Accumulators.AccGeometryCollection acc, Object value) {
+      if (value == null) {
+        return;
+      }
+      if (value instanceof Geography) {
+        Geography geography = (Geography) value;
+        requireType(acc, true);
+        requireMatchingSrid(acc, geography.getSRID());
+        try {
+          acc.values.add(GeographyWKBSerializer.serialize(geography));
+        } catch (IOException e) {
+          throw new IllegalStateException("Failed to serialize Geography", e);
+        }
+      } else if (value instanceof Geometry) {
+        requireType(acc, false);
+        acc.values.add(
+            org.apache.sedona.common.geometrySerde.GeometrySerializer.serialize((Geometry) value));
+      } else {
+        throw new IllegalArgumentException(
+            "ST_Collect_Aggr only accepts Geometry or Geography values");
+      }
+    }
+
+    public void retract(Accumulators.AccGeometryCollection acc, Object value) {
+      if (value == null || acc.values.isEmpty()) {
+        return;
+      }
+
+      byte[] serialized;
+      if (value instanceof Geography) {
+        requireType(acc, true);
+        requireMatchingSrid(acc, ((Geography) value).getSRID());
+        try {
+          serialized = GeographyWKBSerializer.serialize((Geography) value);
+        } catch (IOException e) {
+          throw new IllegalStateException("Failed to serialize Geography", e);
+        }
+      } else if (value instanceof Geometry) {
+        requireType(acc, false);
+        serialized =
+            org.apache.sedona.common.geometrySerde.GeometrySerializer.serialize((Geometry) value);
+      } else {
+        throw new IllegalArgumentException(
+            "ST_Collect_Aggr only accepts Geometry or Geography values");
+      }
+
+      for (int i = 0; i < acc.values.size(); i++) {
+        if (Arrays.equals(acc.values.get(i), serialized)) {
+          acc.values.remove(i);
+          break;
+        }
+      }
+      if (acc.values.isEmpty()) {
+        acc.reset();
+      }
+    }
+
+    public void merge(
+        Accumulators.AccGeometryCollection acc,
+        Iterable<Accumulators.AccGeometryCollection> accumulators) {
+      for (Accumulators.AccGeometryCollection other : accumulators) {
+        if (other.values.isEmpty()) {
+          continue;
+        }
+        requireType(acc, Boolean.TRUE.equals(other.geography));
+        if (Boolean.TRUE.equals(other.geography)) {
+          requireMatchingSrid(acc, other.srid);
+        }
+        acc.values.addAll(other.values);
+      }
+    }
+
+    public void resetAccumulator(Accumulators.AccGeometryCollection acc) {
+      acc.reset();
+    }
+
+    private static void requireType(Accumulators.AccGeometryCollection acc, boolean geography) {
+      if (acc.geography == null) {
+        acc.geography = geography;
+      } else if (acc.geography != geography) {
+        throw new IllegalArgumentException(
+            "ST_Collect_Aggr cannot mix Geometry and Geography values");
+      }
+    }
+
+    private static void requireMatchingSrid(Accumulators.AccGeometryCollection acc, int inputSrid) {
+      if (acc.values.isEmpty()) {
+        acc.srid = inputSrid;
+      } else if (acc.srid != inputSrid) {
+        throw new IllegalArgumentException(
+            "ST_Collect_Aggr requires all Geography values to have the same SRID");
+      }
+    }
+  }
+
   // Aliases for *_Aggr functions with *_Agg suffix
   @DataTypeHint(
       value = "RAW",
@@ -390,4 +557,32 @@ public static class ST_Intersection_Agg extends ST_Intersection_Aggr {}
       rawSerializer = GeometryTypeSerializer.class,
       bridgedTo = Geometry.class)
   public static class ST_Union_Agg extends ST_Union_Aggr {}
+
+  @FunctionHint(
+      input =
+          @DataTypeHint(
+              value = "RAW",
+              rawSerializer = GeometryTypeSerializer.class,
+              bridgedTo = Geometry.class),
+      accumulator =
+          @DataTypeHint(value = "RAW", bridgedTo = Accumulators.AccGeometryCollection.class),
+      output =
+          @DataTypeHint(
+              value = "RAW",
+              rawSerializer = GeometryTypeSerializer.class,
+              bridgedTo = Geometry.class))
+  @FunctionHint(
+      input =
+          @DataTypeHint(
+              value = "RAW",
+              rawSerializer = GeographyTypeSerializer.class,
+              bridgedTo = Geography.class),
+      accumulator =
+          @DataTypeHint(value = "RAW", bridgedTo = Accumulators.AccGeometryCollection.class),
+      output =
+          @DataTypeHint(
+              value = "RAW",
+              rawSerializer = GeographyTypeSerializer.class,
+              bridgedTo = Geography.class))
+  public static class ST_Collect_Agg extends ST_Collect_Aggr {}
 }
diff --git a/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java b/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java
index ff47600ea11..8960c3a82e6 100644
--- a/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java
+++ b/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java
@@ -442,6 +442,36 @@ public org.apache.sedona.common.S2Geography.Geography eval(
   }
 
   public static class ST_Collect extends ScalarFunction {
+    @Override
+    public TypeInference getTypeInference(DataTypeFactory typeFactory) {
+      DataType geometryType =
+          DataTypes.RAW(Geometry.class, GeometryTypeSerializer.INSTANCE).nullable();
+      DataType geographyType =
+          DataTypes.RAW(Geography.class, GeographyTypeSerializer.INSTANCE).nullable();
+      DataType geometryArrayType =
+          DataTypes.ARRAY(geometryType).bridgedTo(Geometry[].class).nullable();
+      DataType geographyArrayType =
+          DataTypes.ARRAY(geographyType).bridgedTo(Geography[].class).nullable();
+
+      InputTypeStrategy geometryPair =
+          InputTypeStrategies.explicitSequence(geometryType, geometryType);
+      InputTypeStrategy geographyPair =
+          InputTypeStrategies.explicitSequence(geographyType, geographyType);
+      InputTypeStrategy geometryArray = InputTypeStrategies.explicitSequence(geometryArrayType);
+      InputTypeStrategy geographyArray = InputTypeStrategies.explicitSequence(geographyArrayType);
+      Map<InputTypeStrategy, TypeStrategy> outputs = new LinkedHashMap<>();
+      outputs.put(geometryPair, TypeStrategies.explicit(geometryType));
+      outputs.put(geographyPair, TypeStrategies.explicit(geographyType));
+      outputs.put(geometryArray, TypeStrategies.explicit(geometryType));
+      outputs.put(geographyArray, TypeStrategies.explicit(geographyType));
+
+      return TypeInference.newBuilder()
+          .inputTypeStrategy(
+              InputTypeStrategies.or(geometryPair, geographyPair, geometryArray, geographyArray))
+          .outputTypeStrategy(TypeStrategies.mapping(outputs))
+          .build();
+    }
+
     @DataTypeHint(
         value = "RAW",
         rawSerializer = GeometryTypeSerializer.class,
@@ -463,14 +493,30 @@ public Geometry eval(
       return org.apache.sedona.common.Functions.createMultiGeometry(geoms);
     }
 
+    @DataTypeHint(
+        value = "RAW",
+        rawSerializer = GeographyTypeSerializer.class,
+        bridgedTo = Geography.class)
+    public Geography eval(Geography geog1, Geography geog2) {
+      return org.apache.sedona.common.geography.Functions.createMultiGeography(
+          new Geography[] {geog1, geog2});
+    }
+
     @DataTypeHint(
         value = "RAW",
         rawSerializer = GeometryTypeSerializer.class,
         bridgedTo = Geometry.class)
-    public Geometry eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
-      Geometry[] geoms = (Geometry[]) o;
+    public Geometry eval(Geometry[] geoms) {
       return org.apache.sedona.common.Functions.createMultiGeometry(geoms);
     }
+
+    @DataTypeHint(
+        value = "RAW",
+        rawSerializer = GeographyTypeSerializer.class,
+        bridgedTo = Geography.class)
+    public Geography eval(Geography[] geographies) {
+      return org.apache.sedona.common.geography.Functions.createMultiGeography(geographies);
+    }
   }
 
   public static class ST_CollectionExtract extends ScalarFunction {
@@ -551,6 +597,19 @@ public Geometry eval(
       Geometry geom = (Geometry) o;
       return org.apache.sedona.common.Functions.convexHull(geom);
     }
+
+    @DataTypeHint(
+        value = "RAW",
+        rawSerializer = GeographyTypeSerializer.class,
+        bridgedTo = Geography.class)
+    public Geography eval(
+        @DataTypeHint(
+                value = "RAW",
+                rawSerializer = GeographyTypeSerializer.class,
+                bridgedTo = Geography.class)
+            Geography geog) {
+      return org.apache.sedona.common.geography.Functions.convexHull(geog);
+    }
   }
 
   public static class ST_CrossesDateLine extends ScalarFunction {
diff --git a/flink/src/test/java/org/apache/sedona/flink/AggregatorTest.java b/flink/src/test/java/org/apache/sedona/flink/AggregatorTest.java
index 1fb0509a84b..ab0c94d95af 100644
--- a/flink/src/test/java/org/apache/sedona/flink/AggregatorTest.java
+++ b/flink/src/test/java/org/apache/sedona/flink/AggregatorTest.java
@@ -21,9 +21,13 @@
 import static org.apache.flink.table.api.Expressions.*;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import org.apache.flink.table.api.*;
 import org.apache.flink.types.Row;
+import org.apache.sedona.common.S2Geography.Geography;
+import org.apache.sedona.common.geography.Constructors;
 import org.apache.sedona.common.geometryObjects.Box2D;
 import org.apache.sedona.common.geometryObjects.Box3D;
 import org.apache.sedona.flink.expressions.Functions;
@@ -202,4 +206,89 @@ public void testUnion_Agg_Alias() {
     Row last = last(result);
     assertEquals(1001, ((Polygon) last.getField(0)).getArea(), 0);
   }
+
+  @Test
+  public void testCollect_Aggr_GeometryAndAlias() {
+    tableEnv.executeSql(
+        "CREATE OR REPLACE TEMPORARY VIEW collect_geom_view AS "
+            + "SELECT ST_GeomFromWKT(wkt) AS geom FROM ("
+            + "VALUES ('POINT (1 2)'), (CAST(NULL AS STRING)), "
+            + "('POINT (1 2)'), ('POINT (3 4)')) AS t(wkt)");
+    Row result =
+        last(
+            tableEnv.sqlQuery(
+                "SELECT ST_Collect_Aggr(geom), ST_Collect_Agg(geom) FROM collect_geom_view"));
+    assertEquals("MULTIPOINT ((1 2), (1 2), (3 4))", result.getField(0).toString());
+    assertEquals(result.getField(0).toString(), result.getField(1).toString());
+  }
+
+  @Test
+  public void testCollect_Agg_GeographyFeedsConvexHull() throws Exception {
+    tableEnv.executeSql(
+        "CREATE OR REPLACE TEMPORARY VIEW collect_geog_view AS "
+            + "SELECT CASE WHEN wkt IS NULL THEN NULL "
+            + "ELSE ST_GeogFromWKT(wkt, 4326) END AS geog FROM ("
+            + "VALUES ('POINT (170 10)'), (CAST(NULL AS STRING)), "
+            + "('POINT (-170 10)'), ('POINT (180 30)'), ('POINT (170 10)')) AS t(wkt)");
+
+    Row result =
+        last(
+            tableEnv.sqlQuery(
+                "SELECT ST_Collect_Aggr(geog), ST_Collect_Agg(geog), "
+                    + "ST_ConvexHull(ST_Collect_Agg(geog)) FROM collect_geog_view"));
+    Geography expectedCollection =
+        org.apache.sedona.common.geography.Functions.createMultiGeography(
+            new Geography[] {
+              Constructors.geogFromWKT("POINT (170 10)", 4326),
+              Constructors.geogFromWKT("POINT (-170 10)", 4326),
+              Constructors.geogFromWKT("POINT (180 30)", 4326),
+              Constructors.geogFromWKT("POINT (170 10)", 4326)
+            });
+    Geography expectedHull =
+        org.apache.sedona.common.geography.Functions.convexHull(expectedCollection);
+
+    assertTrue(
+        Constructors.geogToGeometry(expectedCollection)
+            .equalsNorm(Constructors.geogToGeometry((Geography) result.getField(0))));
+    assertTrue(
+        Constructors.geogToGeometry(expectedCollection)
+            .equalsNorm(Constructors.geogToGeometry((Geography) result.getField(1))));
+    assertEquals(expectedHull.toEWKT(), ((Geography) result.getField(2)).toEWKT());
+  }
+
+  @Test
+  public void testCollect_Agg_AllNullReturnsNull() {
+    tableEnv.executeSql(
+        "CREATE OR REPLACE TEMPORARY VIEW collect_null_geog_view AS "
+            + "SELECT CASE WHEN wkt IS NULL THEN NULL "
+            + "ELSE ST_GeogFromWKT(wkt, 4326) END AS geog FROM ("
+            + "VALUES (CAST(NULL AS STRING)), (CAST(NULL AS STRING))) AS t(wkt)");
+    Row result = last(tableEnv.sqlQuery("SELECT ST_Collect_Agg(geog) FROM collect_null_geog_view"));
+    assertNull(result.getField(0));
+  }
+
+  @Test
+  public void testCollect_Agg_RejectsMixedGeographySrids() {
+    tableEnv.executeSql(
+        "CREATE OR REPLACE TEMPORARY VIEW collect_mixed_srid_view AS "
+            + "SELECT ST_GeogFromWKT(wkt, srid) AS geog FROM ("
+            + "VALUES ('POINT (1 2)', 4326), ('POINT (3 4)', 3857)) AS t(wkt, srid)");
+    try {
+      last(tableEnv.sqlQuery("SELECT ST_Collect_Agg(geog) FROM collect_mixed_srid_view"));
+      fail("Expected ST_Collect_Agg to reject mixed Geography SRIDs");
+    } catch (Exception e) {
+      String messages = messageChain(e);
+      assertTrue(
+          "Expected a mixed-SRID error, got: " + messages,
+          messages.contains("requires all Geography values to have the same SRID"));
+    }
+  }
+
+  private static String messageChain(Throwable t) {
+    StringBuilder sb = new StringBuilder();
+    for (Throwable c = t; c != null; c = c.getCause()) {
+      sb.append(c.getMessage()).append(" | ");
+    }
+    return sb.toString();
+  }
 }
diff --git a/flink/src/test/java/org/apache/sedona/flink/CollectAggregatorTest.java b/flink/src/test/java/org/apache/sedona/flink/CollectAggregatorTest.java
new file mode 100644
index 00000000000..45b1b5e220d
--- /dev/null
+++ b/flink/src/test/java/org/apache/sedona/flink/CollectAggregatorTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.sedona.flink;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import org.apache.sedona.common.S2Geography.Geography;
+import org.apache.sedona.common.geography.Constructors;
+import org.apache.sedona.flink.expressions.Accumulators;
+import org.apache.sedona.flink.expressions.Aggregators;
+import org.junit.Test;
+
+public class CollectAggregatorTest {
+
+  private final Aggregators.ST_Collect_Aggr aggregate = new Aggregators.ST_Collect_Aggr();
+
+  @Test
+  public void testMergeGeographyAccumulators() throws Exception {
+    Geography first = Constructors.geogFromWKT("POINT (170 10)", 4326);
+    Geography second = Constructors.geogFromWKT("POINT (-170 10)", 4326);
+
+    Accumulators.AccGeometryCollection target = aggregate.createAccumulator();
+    aggregate.accumulate(target, first);
+
+    Accumulators.AccGeometryCollection empty = aggregate.createAccumulator();
+    Accumulators.AccGeometryCollection other = aggregate.createAccumulator();
+    aggregate.accumulate(other, second);
+    aggregate.accumulate(other, first);
+
+    aggregate.merge(target, Arrays.asList(empty, other));
+
+    Geography actual = (Geography) aggregate.getValue(target);
+    Geography expected =
+        org.apache.sedona.common.geography.Functions.createMultiGeography(
+            new Geography[] {first, second, first});
+    assertEquals(expected.toEWKT(), actual.toEWKT());
+    assertEquals(3, org.apache.sedona.common.geography.Functions.numGeometries(actual));
+    assertEquals(4326, actual.getSRID());
+  }
+
+  @Test
+  public void testMixedSridRejectedByAccumulateAndMerge() throws Exception {
+    Geography srid4326 = Constructors.geogFromWKT("POINT (1 2)", 4326);
+    Geography srid3857 = Constructors.geogFromWKT("POINT (3 4)", 3857);
+
+    Accumulators.AccGeometryCollection target = aggregate.createAccumulator();
+    aggregate.accumulate(target, srid4326);
+
+    IllegalArgumentException accumulateError =
+        assertThrows(IllegalArgumentException.class, () -> aggregate.accumulate(target, srid3857));
+    assertTrue(accumulateError.getMessage().contains("same SRID"));
+    assertEquals(1, target.values.size());
+
+    Accumulators.AccGeometryCollection other = aggregate.createAccumulator();
+    aggregate.accumulate(other, srid3857);
+    IllegalArgumentException mergeError =
+        assertThrows(
+            IllegalArgumentException.class, () -> aggregate.merge(target, Arrays.asList(other)));
+    assertTrue(mergeError.getMessage().contains("same SRID"));
+    assertEquals(1, target.values.size());
+  }
+
+  @Test
+  public void testRetractAndResetAccumulator() throws Exception {
+    Geography first = Constructors.geogFromWKT("POINT (1 2)", 4326);
+    Geography second = Constructors.geogFromWKT("POINT (3 4)", 4326);
+
+    Accumulators.AccGeometryCollection accumulator = aggregate.createAccumulator();
+    aggregate.accumulate(accumulator, first);
+    aggregate.accumulate(accumulator, first);
+    aggregate.accumulate(accumulator, second);
+
+    aggregate.retract(accumulator, first);
+
+    Geography actual = (Geography) aggregate.getValue(accumulator);
+    Geography expected =
+        org.apache.sedona.common.geography.Functions.createMultiGeography(
+            new Geography[] {first, second});
+    assertEquals(expected.toEWKT(), actual.toEWKT());
+
+    aggregate.resetAccumulator(accumulator);
+    assertNull(aggregate.getValue(accumulator));
+    assertTrue(accumulator.values.isEmpty());
+    assertNull(accumulator.geography);
+    assertEquals(0, accumulator.srid);
+
+    Geography resetSrid = Constructors.geogFromWKT("POINT (5 6)", 3857);
+    aggregate.accumulate(accumulator, resetSrid);
+    assertEquals(3857, ((Geography) aggregate.getValue(accumulator)).getSRID());
+  }
+}
diff --git a/flink/src/test/java/org/apache/sedona/flink/GeographyFunctionTest.java b/flink/src/test/java/org/apache/sedona/flink/GeographyFunctionTest.java
index 05d8d376c4f..5e4bd157ac0 100644
--- a/flink/src/test/java/org/apache/sedona/flink/GeographyFunctionTest.java
+++ b/flink/src/test/java/org/apache/sedona/flink/GeographyFunctionTest.java
@@ -359,6 +359,74 @@ private static String messageChain(Throwable t) {
     return sb.toString();
   }
 
+  @Test
+  public void testConvexHull() throws Exception {
+    String wkt = "MULTIPOINT ((170 10), (-170 10), (180 30), (175 15))";
+    Object out = eval(wkt, call(Functions.ST_ConvexHull.class.getSimpleName(), $("geog")));
+    Geography expected =
+        org.apache.sedona.common.geography.Functions.convexHull(
+            Constructors.geogFromWKT(wkt, 4326));
+    assertEquals(expected.toEWKT(), ((Geography) out).toEWKT());
+    assertEquals(4326, ((Geography) out).getSRID());
+
+    Object empty =
+        eval("LINESTRING EMPTY", call(Functions.ST_ConvexHull.class.getSimpleName(), $("geog")));
+    assertEquals("LINESTRING EMPTY", ((Geography) empty).toString());
+    assertEquals(4326, ((Geography) empty).getSRID());
+  }
+
+  @Test
+  public void testCollectWithTwoInputs() throws Exception {
+    Table geographies =
+        tableEnv.sqlQuery(
+            "SELECT ST_GeogFromWKT('POINT (1 2)', 4326) AS g1, "
+                + "ST_GeogFromWKT('POINT (-2 3)', 4326) AS g2");
+    Geography actual =
+        (Geography)
+            first(
+                    geographies.select(
+                        call(Functions.ST_Collect.class.getSimpleName(), $("g1"), $("g2"))))
+                .getField(0);
+    Geography expected =
+        org.apache.sedona.common.geography.Functions.createMultiGeography(
+            new Geography[] {
+              Constructors.geogFromWKT("POINT (1 2)", 4326),
+              Constructors.geogFromWKT("POINT (-2 3)", 4326)
+            });
+    assertEquals(expected.toEWKT(), actual.toEWKT());
+  }
+
+  @Test
+  public void testCollectWithArray() throws Exception {
+    Table geographies =
+        tableEnv.sqlQuery(
+            "SELECT ARRAY["
+                + "ST_GeogFromWKT('LINESTRING (1 2, 3 4)', 4326), "
+                + "ST_GeogFromWKT('LINESTRING (3 4, 4 5)', 4326)] AS geogs");
+    Geography actual =
+        (Geography)
+            first(geographies.select(call(Functions.ST_Collect.class.getSimpleName(), $("geogs"))))
+                .getField(0);
+    Geography expected =
+        org.apache.sedona.common.geography.Functions.createMultiGeography(
+            new Geography[] {
+              Constructors.geogFromWKT("LINESTRING (1 2, 3 4)", 4326),
+              Constructors.geogFromWKT("LINESTRING (3 4, 4 5)", 4326)
+            });
+    assertEquals(expected.toEWKT(), actual.toEWKT());
+
+    Table withEmpty =
+        tableEnv.sqlQuery(
+            "SELECT ARRAY["
+                + "ST_GeogFromWKT('LINESTRING EMPTY', 4326), "
+                + "ST_GeogFromWKT('LINESTRING (3 4, 4 5)', 4326)] AS geogs");
+    Geography actualWithEmpty =
+        (Geography)
+            first(withEmpty.select(call(Functions.ST_Collect.class.getSimpleName(), $("geogs"))))
+                .getField(0);
+    assertEquals(2, org.apache.sedona.common.geography.Functions.numGeometries(actualWithEmpty));
+  }
+
   @Test
   public void testGeometryStillWorks() throws Exception {
     // The geometry overload must remain selectable on the same function.
diff --git a/spark/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala b/spark/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala
index bdb1335bf7c..895b03ff22a 100644
--- a/spark/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala
+++ b/spark/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala
@@ -326,6 +326,9 @@ object Catalog extends AbstractCatalog with Logging {
   // Other / utility expressions not in any docs category
   val otherExprs: Seq[FunctionDescription] = Seq(function[Barrier]())
 
+  // Aggregate functions implemented as native Catalyst expressions
+  val catalystAggregateExprs: Seq[FunctionDescription] = Seq(function[ST_Collect_Agg]())
+
   // Geography (ST_Geog*) — see docs/api/sql/geography/Geography-Functions
   val geographyExprs: Seq[FunctionDescription] = Seq(
     function[ST_GeogFromWKT](0),
@@ -527,6 +530,7 @@ object Catalog extends AbstractCatalog with Logging {
     spatialIndexingExprs,
     addressExprs,
     otherExprs,
+    catalystAggregateExprs,
     geographyExprs,
     rasterConstructorExprs,
     rasterAccessorExprs,
@@ -553,6 +557,5 @@ object Catalog extends AbstractCatalog with Logging {
       new ST_Extent,
       new ST_3DExtent,
       new ST_Intersection_Aggr,
-      new ST_Union_Aggr(),
-      new ST_Collect_Agg())
+      new ST_Union_Aggr())
 }
diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/AggregateFunctions.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/AggregateFunctions.scala
index f7d73740c4b..55d9fcdaadf 100644
--- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/AggregateFunctions.scala
+++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/AggregateFunctions.scala
@@ -317,42 +317,3 @@ private[apache] class ST_Intersection_Aggr
 
   override def finish(out: Geometry): Geometry = out
 }
-
-/**
- * Return a multi-geometry collection of all geometries in the given column. Unlike ST_Union_Aggr,
- * this function does not dissolve boundaries between geometries.
- */
-private[apache] class ST_Collect_Agg
-    extends Aggregator[Geometry, ListBuffer[Geometry], Geometry] {
-
-  val serde = ExpressionEncoder[Geometry]()
-  val bufferSerde = ExpressionEncoder[ListBuffer[Geometry]]()
-
-  override def reduce(buffer: ListBuffer[Geometry], input: Geometry): ListBuffer[Geometry] = {
-    if (input != null) {
-      buffer += input
-    }
-    buffer
-  }
-
-  override def merge(
-      buffer1: ListBuffer[Geometry],
-      buffer2: ListBuffer[Geometry]): ListBuffer[Geometry] = {
-    buffer1 ++= buffer2
-    buffer1
-  }
-
-  override def finish(reduction: ListBuffer[Geometry]): Geometry = {
-    if (reduction.isEmpty) {
-      null
-    } else {
-      Functions.createMultiGeometry(reduction.toArray)
-    }
-  }
-
-  def bufferEncoder: ExpressionEncoder[ListBuffer[Geometry]] = bufferSerde
-
-  def outputEncoder: ExpressionEncoder[Geometry] = serde
-
-  override def zero: ListBuffer[Geometry] = ListBuffer.empty
-}
diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Functions.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Functions.scala
index 3a3ea58ccfb..79640e00076 100644
--- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Functions.scala
+++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Functions.scala
@@ -152,7 +152,9 @@ private[apache] case class ST_ConcaveHull(inputExpressions: Seq[Expression])
  * @param inputExpressions
  */
 private[apache] case class ST_ConvexHull(inputExpressions: Seq[Expression])
-    extends InferredExpression(Functions.convexHull _) {
+    extends InferredExpression(
+      inferrableFunction1(Functions.convexHull),
+      inferrableFunction1(org.apache.sedona.common.geography.Functions.convexHull)) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)
diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/ST_CollectAgg.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/ST_CollectAgg.scala
new file mode 100644
index 00000000000..097dd5ef324
--- /dev/null
+++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/ST_CollectAgg.scala
@@ -0,0 +1,185 @@
+/*
+ * 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.spark.sql.sedona_sql.expressions
+
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, DataInputStream, DataOutputStream}
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.sedona.common.Functions
+import org.apache.sedona.common.S2Geography.GeographyWKBSerializer
+import org.apache.sedona.common.geography.{Functions => GeographyFunctions}
+import org.apache.sedona.sql.utils.GeometrySerializer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Expression, ImplicitCastInputTypes}
+import org.apache.spark.sql.catalyst.expressions.aggregate.{ImperativeAggregate, TypedImperativeAggregate}
+import org.apache.spark.sql.catalyst.trees.UnaryLike
+import org.apache.spark.sql.sedona_sql.UDT.{GeographyUDT, GeometryUDT}
+import org.apache.spark.sql.types.{AbstractDataType, DataType, TypeCollection}
+
+/**
+ * Collect all non-null spatial values in a column without dissolving boundaries.
+ *
+ * Unlike a typed [[org.apache.spark.sql.expressions.Aggregator]], this Catalyst aggregate can
+ * preserve the logical type of its child: Geometry inputs produce Geometry and Geography inputs
+ * produce Geography. The aggregation buffer retains the already-serialized UDT values so it can
+ * be shuffled without choosing one UDT encoder for both overloads.
+ */
+private[apache] case class ST_Collect_Agg(
+    child: Expression,
+    mutableAggBufferOffset: Int = 0,
+    inputAggBufferOffset: Int = 0)
+    extends TypedImperativeAggregate[ArrayBuffer[Array[Byte]]]
+    with ImplicitCastInputTypes
+    with UnaryLike[Expression] {
+
+  def this(inputExpressions: Seq[Expression]) =
+    this(ST_Collect_Agg.requireSingleInput(inputExpressions), 0, 0)
+
+  override def nullable: Boolean = true
+
+  override def dataType: DataType = child.dataType match {
+    case _: GeographyUDT => GeographyUDT()
+    case _ => GeometryUDT()
+  }
+
+  override def inputTypes: Seq[AbstractDataType] =
+    Seq(TypeCollection(GeometryUDT(), GeographyUDT()))
+
+  private def isGeography: Boolean = child.dataType.isInstanceOf[GeographyUDT]
+
+  override def createAggregationBuffer(): ArrayBuffer[Array[Byte]] = ArrayBuffer.empty
+
+  override def update(
+      buffer: ArrayBuffer[Array[Byte]],
+      input: InternalRow): ArrayBuffer[Array[Byte]] = {
+    val value = child.eval(input).asInstanceOf[Array[Byte]]
+    if (value != null) {
+      if (isGeography && buffer.nonEmpty) {
+        ST_Collect_Agg.requireMatchingSRID(buffer.head, value)
+      }
+      // Catalyst can reuse input-row storage, so retain an owned copy in the aggregation buffer.
+      buffer += value.clone()
+    }
+    buffer
+  }
+
+  override def merge(
+      buffer: ArrayBuffer[Array[Byte]],
+      other: ArrayBuffer[Array[Byte]]): ArrayBuffer[Array[Byte]] = {
+    if (isGeography && buffer.nonEmpty && other.nonEmpty) {
+      ST_Collect_Agg.requireMatchingSRID(buffer.head, other.head)
+    }
+    buffer ++= other
+  }
+
+  override def eval(buffer: ArrayBuffer[Array[Byte]]): Any = {
+    if (buffer.isEmpty) {
+      null
+    } else if (isGeography) {
+      val geographies = buffer.map(GeographyWKBSerializer.deserialize)
+      GeographyWKBSerializer.serialize(
+        GeographyFunctions.createMultiGeography(geographies.toArray))
+    } else {
+      val geometries = buffer.map(GeometrySerializer.deserialize)
+      GeometrySerializer.serialize(Functions.createMultiGeometry(geometries.toArray))
+    }
+  }
+
+  override def serialize(buffer: ArrayBuffer[Array[Byte]]): Array[Byte] = {
+    val bytes = new ByteArrayOutputStream()
+    val output = new DataOutputStream(bytes)
+    output.writeInt(buffer.size)
+    buffer.foreach { value =>
+      output.writeInt(value.length)
+      output.write(value)
+    }
+    output.flush()
+    bytes.toByteArray
+  }
+
+  override def deserialize(storageFormat: Array[Byte]): ArrayBuffer[Array[Byte]] = {
+    if (storageFormat == null) {
+      return createAggregationBuffer()
+    }
+
+    val input = new DataInputStream(new ByteArrayInputStream(storageFormat))
+    val count = input.readInt()
+    if (count < 0) {
+      throw new IllegalArgumentException(s"Invalid ST_Collect_Agg buffer count: $count")
+    }
+
+    val buffer = new ArrayBuffer[Array[Byte]](count)
+    var index = 0
+    while (index < count) {
+      val length = input.readInt()
+      if (length < 0 || length > input.available()) {
+        throw new IllegalArgumentException(s"Invalid ST_Collect_Agg buffer item length: $length")
+      }
+      val value = new Array[Byte](length)
+      input.readFully(value)
+      buffer += value
+      index += 1
+    }
+    buffer
+  }
+
+  override def withNewMutableAggBufferOffset(
+      newMutableAggBufferOffset: Int): ImperativeAggregate =
+    copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+  override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =
+    copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+  override protected def withNewChildInternal(newChild: Expression): ST_Collect_Agg =
+    copy(child = newChild)
+
+  override def prettyName: String = "st_collect_agg"
+}
+
+private object ST_Collect_Agg {
+
+  private def requireSingleInput(inputExpressions: Seq[Expression]): Expression = {
+    if (inputExpressions.length != 1) {
+      throw new IllegalArgumentException(
+        s"ST_Collect_Agg requires exactly one argument, but got ${inputExpressions.length}")
+    }
+    inputExpressions.head
+  }
+
+  private def readGeographySRID(value: Array[Byte]): Int = {
+    if (value.length < Integer.BYTES) {
+      throw new IllegalArgumentException("Invalid serialized Geography in ST_Collect_Agg")
+    }
+    ((value(0) & 0xff) << 24) |
+      ((value(1) & 0xff) << 16) |
+      ((value(2) & 0xff) << 8) |
+      (value(3) & 0xff)
+  }
+
+  private def requireMatchingSRID(left: Array[Byte], right: Array[Byte]): Unit = {
+    val leftSRID = readGeographySRID(left)
+    val rightSRID = readGeographySRID(right)
+    if (leftSRID != rightSRID) {
+      throw new IllegalArgumentException(
+        s"ST_Collect_Agg requires all Geography inputs to have the same SRID; " +
+          s"found $leftSRID and $rightSRID")
+    }
+  }
+}
diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/collect/ST_Collect.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/collect/ST_Collect.scala
index 4892fb5a8b3..cb3fad235a6 100644
--- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/collect/ST_Collect.scala
+++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/collect/ST_Collect.scala
@@ -19,11 +19,14 @@
 package org.apache.spark.sql.sedona_sql.expressions.collect
 
 import org.apache.sedona.common.Functions
+import org.apache.sedona.common.S2Geography.Geography
 import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.catalyst.util.ArrayData
-import org.apache.spark.sql.sedona_sql.UDT.GeometryUDT
+import org.apache.spark.sql.sedona_sql.UDT.{GeographyUDT, GeometryUDT}
 import org.apache.spark.sql.sedona_sql.expressions.implicits._
 import org.apache.spark.sql.sedona_sql.expressions.{InferredExpression, SerdeAware}
 import org.apache.spark.sql.types.{ArrayType, _}
@@ -38,51 +41,107 @@ private[apache] case class ST_Collect(inputExpressions: Seq[Expression])
   override def nullable: Boolean = true
 
   override def eval(input: InternalRow): Any = {
-    evalWithoutSerialization(input).asInstanceOf[Geometry].toGenericArrayData
+    val result = evalWithoutSerialization(input)
+    if (result == null) {
+      null
+    } else if (returnsGeography) {
+      result.asInstanceOf[Geography].toGenericArrayData
+    } else {
+      result.asInstanceOf[Geometry].toGenericArrayData
+    }
   }
 
   override def evalWithoutSerialization(input: InternalRow): Any = {
     val firstElement = inputExpressions.head
 
-    firstElement.dataType match {
-      case ArrayType(elementType, _) =>
-        elementType match {
-          case _: GeometryUDT =>
-            val data = firstElement.eval(input).asInstanceOf[ArrayData]
-            val numElements = data.numElements()
-            val geomElements = (0 until numElements)
-              .map(element => data.getBinary(element))
-              .filter(_ != null)
-              .map(_.toGeometry)
-
-            try {
-              Functions.createMultiGeometry(geomElements.toArray)
-            } catch {
-              case e: Exception =>
-                InferredExpression.throwExpressionInferenceException(
-                  getClass.getSimpleName,
-                  Seq(geomElements),
-                  e)
-            }
-
-          case _ => Functions.createMultiGeometry(Array())
-        }
-      case _ =>
-        val geomElements =
+    if (returnsGeography) {
+      val geographies = firstElement.dataType match {
+        case ArrayType(_, _) =>
+          Option(firstElement.eval(input).asInstanceOf[ArrayData])
+            .map(data =>
+              (0 until data.numElements())
+                .filterNot(data.isNullAt)
+                .map(element => data.getBinary(element).toGeography))
+            .getOrElse(Seq.empty)
+        case _ =>
+          inputExpressions.map(_.toGeography(input)).filter(_ != null)
+      }
+      try {
+        org.apache.sedona.common.geography.Functions.createMultiGeography(geographies.toArray)
+      } catch {
+        case e: Exception =>
+          InferredExpression.throwExpressionInferenceException(
+            getClass.getSimpleName,
+            Seq(geographies),
+            e)
+      }
+    } else {
+      val geometries = firstElement.dataType match {
+        case ArrayType(_, _) =>
+          Option(firstElement.eval(input).asInstanceOf[ArrayData])
+            .map(data =>
+              (0 until data.numElements())
+                .filterNot(data.isNullAt)
+                .map(element => data.getBinary(element).toGeometry))
+            .getOrElse(Seq.empty)
+        case _ =>
           inputExpressions.map(_.toGeometry(input)).filter(_ != null)
-        try {
-          Functions.createMultiGeometry(geomElements.toArray)
-        } catch {
-          case e: Exception =>
-            InferredExpression.throwExpressionInferenceException(
-              getClass.getSimpleName,
-              Seq(geomElements),
-              e)
-        }
+      }
+      try {
+        Functions.createMultiGeometry(geometries.toArray)
+      } catch {
+        case e: Exception =>
+          InferredExpression.throwExpressionInferenceException(
+            getClass.getSimpleName,
+            Seq(geometries),
+            e)
+      }
     }
   }
 
-  override def dataType: DataType = GeometryUDT()
+  private def elementType(dataType: DataType): DataType = dataType match {
+    case ArrayType(element, _) => element
+    case other => other
+  }
+
+  private def isGeometry(dataType: DataType): Boolean =
+    dataType.isInstanceOf[GeometryUDT]
+
+  private def isGeography(dataType: DataType): Boolean =
+    dataType.isInstanceOf[GeographyUDT]
+
+  private def returnsGeography: Boolean =
+    inputExpressions.exists(expression => isGeography(elementType(expression.dataType)))
+
+  override def checkInputDataTypes(): TypeCheckResult = {
+    val hasArray = inputExpressions.exists(_.dataType.isInstanceOf[ArrayType])
+    if (hasArray && inputExpressions.length != 1) {
+      return TypeCheckFailure("ST_Collect accepts either one array or one or more scalar values")
+    }
+
+    val unsupported = inputExpressions.map(_.dataType).filterNot { dataType =>
+      val valueType = elementType(dataType)
+      valueType == NullType || isGeometry(valueType) || isGeography(valueType)
+    }
+    if (unsupported.nonEmpty) {
+      return TypeCheckFailure(
+        s"ST_Collect expects Geometry or Geography values, but found ${unsupported.mkString(", ")}")
+    }
+
+    val hasGeometry =
+      inputExpressions.exists(expression => isGeometry(elementType(expression.dataType)))
+    val hasGeography =
+      inputExpressions.exists(expression => isGeography(elementType(expression.dataType)))
+    if (hasGeometry && hasGeography) {
+      TypeCheckFailure("ST_Collect does not accept mixed Geometry and Geography inputs")
+    } else {
+      TypeCheckSuccess
+    }
+  }
+
+  override def dataType: DataType = {
+    if (returnsGeography) GeographyUDT() else GeometryUDT()
+  }
 
   override def children: Seq[Expression] = inputExpressions
 
diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_aggregates.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_aggregates.scala
index 422165f1be5..560ede6b826 100644
--- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_aggregates.scala
+++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_aggregates.scala
@@ -20,7 +20,7 @@ package org.apache.spark.sql.sedona_sql.expressions
 
 import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.Column
-import org.apache.spark.sql.functions.{col, udaf}
+import org.apache.spark.sql.functions.{call_udf, col, udaf}
 
 object st_aggregates {
   def ST_Envelope_Aggr(geometry: Column): Column = {
@@ -54,13 +54,11 @@ object st_aggregates {
   }
 
   def ST_Collect_Agg(geometry: Column): Column = {
-    val aggrFunc = udaf(new ST_Collect_Agg)
-    aggrFunc(geometry)
+    call_udf("ST_Collect_Agg", geometry)
   }
 
   def ST_Collect_Agg(geometry: String): Column = {
-    val aggrFunc = udaf(new ST_Collect_Agg)
-    aggrFunc(col(geometry))
+    ST_Collect_Agg(col(geometry))
   }
 
   def ST_Extent(geometry: Column): Column = {
diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/geography/GeographyFunctionTest.scala b/spark/common/src/test/scala/org/apache/sedona/sql/geography/GeographyFunctionTest.scala
index 7899c8c5a1c..b562dd73e85 100644
--- a/spark/common/src/test/scala/org/apache/sedona/sql/geography/GeographyFunctionTest.scala
+++ b/spark/common/src/test/scala/org/apache/sedona/sql/geography/GeographyFunctionTest.scala
@@ -21,6 +21,7 @@ package org.apache.sedona.sql.geography
 import org.apache.sedona.common.S2Geography.{Geography, WKBGeography}
 import org.apache.sedona.sql.TestBaseScala
 import org.apache.spark.sql.functions.{col, lit}
+import org.apache.spark.sql.sedona_sql.UDT.GeographyUDT
 import org.apache.spark.sql.sedona_sql.expressions.{st_constructors, st_functions, st_predicates}
 import org.junit.Assert.{assertEquals, assertFalse, assertNotNull, assertTrue}
 import org.locationtech.jts.geom.Point
@@ -187,6 +188,188 @@ class GeographyFunctionTest extends TestBaseScala {
       assertTrue(row.isNullAt(1))
     }
 
+    it("ST_ConvexHull uses spherical Geography semantics") {
+      val pointHull = sparkSession
+        .sql("SELECT ST_ConvexHull(ST_GeogFromWKT('MULTIPOINT ((1 2), (1 2))', 4326)) AS hull")
+      assertTrue(pointHull.schema("hull").dataType.isInstanceOf[GeographyUDT])
+      assertEquals("POINT (1 2)", pointHull.first().getAs[Geography](0).toString)
+
+      val lineHull = sparkSession
+        .sql("SELECT ST_ConvexHull(ST_GeogFromWKT('LINESTRING (0 0, 0 1, 0 2)', 4326)) AS hull")
+        .first()
+        .getAs[Geography](0)
+      assertEquals(
+        "ST_LineString",
+        org.apache.sedona.common.geography.Functions.geometryType(lineHull))
+      assertEquals(4326, lineHull.getSRID)
+
+      val polygonHull = sparkSession
+        .sql("""
+          SELECT ST_ConvexHull(
+            ST_GeogFromWKT(
+              'MULTIPOINT ((170 -10), (170 10), (-170 10), (-170 -10))',
+              4326
+            )
+          ) AS hull
+        """)
+        .first()
+        .getAs[Geography](0)
+      assertEquals(
+        "ST_Polygon",
+        org.apache.sedona.common.geography.Functions.geometryType(polygonHull))
+      assertTrue(org.apache.sedona.common.geography.Functions.area(polygonHull) < 1e14)
+
+      val emptyLineHull = sparkSession
+        .sql("SELECT ST_ConvexHull(ST_GeogFromWKT('LINESTRING EMPTY', 4326)) AS hull")
+        .first()
+        .getAs[Geography](0)
+      assertEquals("LINESTRING EMPTY", emptyLineHull.toString)
+      assertEquals(4326, emptyLineHull.getSRID)
+    }
+
+    it("ST_Collect accepts Geography arrays and scalar arguments") {
+      val arrayResult = sparkSession.sql("""
+        SELECT ST_Collect(array(
+          ST_GeogFromWKT('POINT (1 2)', 4326),
+          ST_GeogFromWKT(NULL, 4326),
+          ST_GeogFromWKT('POINT (3 4)', 4326)
+        )) AS collected
+      """)
+      assertTrue(arrayResult.schema("collected").dataType.isInstanceOf[GeographyUDT])
+      assertEquals("MULTIPOINT ((1 2), (3 4))", arrayResult.first().getAs[Geography](0).toString)
+
+      val mixedShapeResult = sparkSession
+        .sql("""
+          SELECT ST_Collect(
+            ST_GeogFromWKT('POINT (1 2)', 4326),
+            ST_GeogFromWKT('LINESTRING (0 0, 1 1)', 4326)
+          ) AS collected
+        """)
+        .first()
+        .getAs[Geography](0)
+      assertEquals(
+        "ST_GeometryCollection",
+        org.apache.sedona.common.geography.Functions.geometryType(mixedShapeResult))
+      assertEquals(4326, mixedShapeResult.getSRID)
+
+      val withEmptyLine = sparkSession
+        .sql("""
+          SELECT ST_Collect(array(
+            ST_GeogFromWKT('LINESTRING EMPTY', 4326),
+            ST_GeogFromWKT('LINESTRING (0 0, 1 1)', 4326)
+          )) AS collected
+        """)
+        .first()
+        .getAs[Geography](0)
+      assertEquals(2, org.apache.sedona.common.geography.Functions.numGeometries(withEmptyLine))
+    }
+
+    it("ST_Collect rejects mixed scalar and array arguments") {
+      val error = intercept[org.apache.spark.sql.AnalysisException] {
+        sparkSession
+          .sql("""
+            SELECT ST_Collect(
+              ST_GeogFromWKT('POINT (1 2)', 4326),
+              array(ST_GeogFromWKT('POINT (3 4)', 4326))
+            )
+          """)
+          .collect()
+      }
+      assertTrue(error.getMessage.contains("either one array or one or more scalar values"))
+
+      val nestedArrayError = intercept[org.apache.spark.sql.AnalysisException] {
+        sparkSession
+          .sql("""
+            SELECT ST_Collect(
+              array(array(ST_GeogFromWKT('POINT (1 2)', 4326)))
+            )
+          """)
+          .collect()
+      }
+      assertTrue(nestedArrayError.getMessage.contains("expects Geometry or Geography values"))
+    }
+
+    it("computes the grouped Geography hull from ARRAY_AGG") {
+      val result = sparkSession
+        .sql("""
+          WITH dropoffs AS (
+            SELECT * FROM VALUES
+              (1, ST_GeogFromWKT('POINT (0 0)', 4326)),
+              (1, ST_GeogFromWKT('POINT (1 0)', 4326)),
+              (1, ST_GeogFromWKT('POINT (0 1)', 4326))
+            AS dropoffs(customer_id, geog)
+          )
+          SELECT
+            ST_GeometryType(ST_ConvexHull(ST_Collect(ARRAY_AGG(geog)))) AS hull_type,
+            ST_Area(ST_ConvexHull(ST_Collect(ARRAY_AGG(geog)))) AS area
+          FROM dropoffs
+          GROUP BY customer_id
+        """)
+        .first()
+
+      assertEquals("ST_Polygon", result.getString(0))
+      assertTrue(result.getDouble(1) > 6e9)
+    }
+
+    it("ST_Collect_Agg accepts Geography and feeds ST_ConvexHull") {
+      val aggregate = sparkSession.sql("""
+        WITH dropoffs AS (
+          SELECT * FROM VALUES
+            (1, ST_GeogFromWKT('POINT (0 0)', 4326)),
+            (1, ST_GeogFromWKT('POINT (1 0)', 4326)),
+            (1, ST_GeogFromWKT('POINT (0 1)', 4326))
+          AS dropoffs(customer_id, geog)
+        )
+        SELECT
+          ST_Collect_Agg(geog) AS collected,
+          ST_Area(ST_ConvexHull(ST_Collect_Agg(geog))) AS hull_area
+        FROM dropoffs
+        GROUP BY customer_id
+      """)
+
+      assertTrue(aggregate.schema("collected").dataType.isInstanceOf[GeographyUDT])
+      val row = aggregate.first()
+      val collected = row.getAs[Geography]("collected")
+      assertEquals(3, org.apache.sedona.common.geography.Functions.numGeometries(collected))
+      assertEquals(4326, collected.getSRID)
+      assertTrue(row.getAs[Double]("hull_area") > 6e9)
+
+      val allNull = sparkSession.sql("""
+        SELECT ST_Collect_Agg(geog) AS collected
+        FROM (
+          SELECT ST_GeogFromWKT(NULL, 4326) AS geog
+          UNION ALL
+          SELECT ST_GeogFromWKT(NULL, 4326) AS geog
+        )
+      """)
+      assertTrue(allNull.schema("collected").dataType.isInstanceOf[GeographyUDT])
+      assertTrue(allNull.first().isNullAt(0))
+    }
+
+    it("ST_Collect_Agg rejects mixed Geography SRIDs") {
+      val error = intercept[Exception] {
+        sparkSession
+          .sql("""
+            SELECT ST_Collect_Agg(geog)
+            FROM (
+              SELECT ST_GeogFromWKT('POINT (0 0)', 4326) AS geog
+              UNION ALL
+              SELECT ST_GeogFromWKT('POINT (1 1)', 3857) AS geog
+            )
+          """)
+          .collect()
+      }
+
+      var cause: Throwable = error
+      var hasExpectedMessage = false
+      while (cause != null) {
+        hasExpectedMessage ||= Option(cause.getMessage)
+          .exists(_.contains("same SRID"))
+        cause = cause.getCause
+      }
+      assertTrue(hasExpectedMessage)
+    }
+
     it("ST_MakeLine creates a geography measured in meters") {
       val row = sparkSession
         .sql("""
diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/geography/PreserveSRIDGeographySuite.scala b/spark/common/src/test/scala/org/apache/sedona/sql/geography/PreserveSRIDGeographySuite.scala
index aa09bf94c04..6506c87adb4 100644
--- a/spark/common/src/test/scala/org/apache/sedona/sql/geography/PreserveSRIDGeographySuite.scala
+++ b/spark/common/src/test/scala/org/apache/sedona/sql/geography/PreserveSRIDGeographySuite.scala
@@ -58,6 +58,9 @@ class PreserveSRIDGeographySuite extends TestBaseScala with TableDrivenPropertyC
       ("ST_Buffer(geog1, 0)", 4326),
       ("ST_Buffer(geog1, 100)", 4326),
       ("ST_Buffer(geog1, 100, 'quad_segs=8')", 4326),
+      ("ST_ConvexHull(geog1)", 4326),
+      ("ST_Collect(geog1, geog2)", 4326),
+      ("ST_Collect(array(geog1, geog2))", 4326),
       ("ST_MakeLine(geog3, geog3)", 4326),
       // Cross-type boundaries. The literal SRID here exercises that any int survives the
       // Geometry↔Geography boundary (no CRS resolution is involved on this code path).