[GH-3203] Preserve Geography coordinates and polygon ring semantics

common/src/main/java/org/apache/sedona/common/S2Geography/WKBGeography.java

@@ -341,11 +341,14 @@ public void getCellUnionBound(List<S2CellId> cellIds) {
 
   @Override
   public String toString() {
-    return getS2Geography().toString();
+    // WKB is the structural source of truth. Rendering the derived S2 view here would both
+    // canonicalize repeated vertices and apply Geography's default fixed-precision formatting.
+    return getJTSGeometry().toText();
   }
 
   @Override
   public String toString(PrecisionModel precisionModel) {
+    // Keep the S2 writer available to callers that explicitly request precision-controlled text.
     return getS2Geography().toString(precisionModel);
   }
 
@@ -356,9 +359,8 @@ public String toText(PrecisionModel precisionModel) {
 
   @Override
   public String toEWKT() {
-    Geography s2 = getS2Geography();
-    s2.setSRID(getSRID());
-    return s2.toEWKT();
+    String wkt = toString();
+    return getSRID() > 0 ? "SRID=" + getSRID() + "; " + wkt : wkt;
   }
 
   @Override

common/src/main/java/org/apache/sedona/common/S2Geography/WKBReader.java

@@ -343,7 +343,15 @@ private PolygonGeography readPolygon(EnumSet<Ordinate> ordinateFlags)
       for (int i = 0; i < seq.size(); i++) {
         pts.add(S2LatLng.fromDegrees(seq.getY(i), seq.getX(i)).toPoint());
       }
-      loops.add(new S2Loop(pts));
+      S2Loop loop = new S2Loop(pts);
+      // Match simple-features/SedonaDB semantics: ring position determines whether it is a shell
+      // or hole. Normalize only the operational S2 loop; WKBGeography retains the source bytes.
+      boolean isHole = r > 0;
+      boolean isClockwise = loop.getTurningAngle() < 0;
+      if (isHole != isClockwise) {
+        loop.invert();
+      }
+      loops.add(loop);
     }
     if (loops.isEmpty()) {
       return new PolygonGeography();

common/src/main/java/org/apache/sedona/common/S2Geography/WKTReader.java

@@ -821,14 +821,16 @@ private PolygonGeography readPolygonText(
       return new PolygonGeography();
     }
 
-    List<S2Loop> holes = new ArrayList<S2Loop>();
+    List<S2Loop> loops = new ArrayList<S2Loop>();
 
     S2Loop shell = readLoopText(tokenizer, ordinateFlags);
-    holes.add(shell);
+    normalizePolygonLoop(shell, false);
+    loops.add(shell);
     nextToken = getNextCloserOrComma(tokenizer);
     while (nextToken.equals(COMMA)) {
       S2Loop hole = readLoopText(tokenizer, ordinateFlags);
-      holes.add(hole);
+      normalizePolygonLoop(hole, true);
+      loops.add(hole);
       nextToken = getNextCloserOrComma(tokenizer);
     }
 
@@ -838,7 +840,7 @@ private PolygonGeography readPolygonText(
     builder.startLayer(polyLayer);
 
     // add shell + holes
-    for (S2Loop loop : holes) {
+    for (S2Loop loop : loops) {
       builder.addLoop(loop);
     }
 
@@ -855,6 +857,18 @@ private PolygonGeography readPolygonText(
     return new PolygonGeography(s2poly);
   }
 
+  /**
+   * Normalizes an operational S2 loop using its simple-features ring role. Structural WKT callers
+   * use {@link org.apache.sedona.common.geography.Constructors}, which keeps the submitted
+   * coordinate order in WKBGeography; this reader produces only the derived S2 representation.
+   */
+  private static void normalizePolygonLoop(S2Loop loop, boolean isHole) {
+    boolean isClockwise = loop.getTurningAngle() < 0;
+    if (isHole != isClockwise) {
+      loop.invert();
+    }
+  }
+
   /**
    * Creates a <code>MultiLineString</code> using the next token in the stream.
    *

common/src/main/java/org/apache/sedona/common/S2Geography/WkbS2Shape.java

@@ -21,11 +21,16 @@
 import com.google.common.geometry.S2;
 import com.google.common.geometry.S2EdgeUtil;
 import com.google.common.geometry.S2LatLng;
+import com.google.common.geometry.S2Loop;
 import com.google.common.geometry.S2Point;
 import com.google.common.geometry.S2Predicates;
 import com.google.common.geometry.S2Shape;
+import com.google.common.geometry.S2ShapeMeasures;
+import com.google.common.geometry.S2ShapeUtil;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * An S2Shape implementation that reads WKB bytes once, converts all coordinates to S2Points in the
@@ -48,11 +53,29 @@ public class WkbS2Shape implements S2Shape {
   private final int[] chainStarts; // edge offset for each chain
   private final int[] chainLengths; // edge count for each chain
   private final int[] vertexOffsets; // index into vertices[] for first vertex of each chain
+  // Polygon rings are normalized virtually: source WKB and vertices[] remain unchanged.
+  private final boolean[] chainReversed;
 
   // For polygon containsOrigin — computed eagerly at construction for polygons
   private final boolean containsOriginValue;
 
   public WkbS2Shape(byte[] wkb) {
+    this(wkb, true);
+  }
+
+  /**
+   * Builds a shape whose polygon rings retain their WKB traversal direction.
+   *
+   * <p>This is an internal escape hatch for callers that intentionally encode the spherical
+   * interior through ring direction, such as raster footprints that cover more than a hemisphere.
+   * General Geography construction must use {@link #WkbS2Shape(byte[])}, which applies
+   * simple-features shell/hole semantics independent of input winding.
+   */
+  public static WkbS2Shape withPreservedLoopOrientation(byte[] wkb) {
+    return new WkbS2Shape(wkb, false);
+  }
+
+  private WkbS2Shape(byte[] wkb, boolean normalizePolygonRings) {
     boolean le = (wkb[0] == 0x01);
     ByteBuffer buf =
         ByteBuffer.wrap(wkb).order(le ? ByteOrder.LITTLE_ENDIAN : ByteOrder.BIG_ENDIAN);
@@ -80,6 +103,7 @@ public WkbS2Shape(byte[] wkb) {
           this.chainStarts = new int[] {0};
           this.chainLengths = new int[] {1};
           this.vertexOffsets = new int[] {0};
+          this.chainReversed = new boolean[] {false};
           this.containsOriginValue = false;
           break;
         }
@@ -93,6 +117,7 @@ public WkbS2Shape(byte[] wkb) {
           this.chainStarts = new int[] {0};
           this.chainLengths = new int[] {totalEdges};
           this.vertexOffsets = new int[] {0};
+          this.chainReversed = new boolean[] {false};
           this.containsOriginValue = false;
           break;
         }
@@ -104,6 +129,7 @@ public WkbS2Shape(byte[] wkb) {
           this.chainStarts = new int[numRings];
           this.chainLengths = new int[numRings];
           this.vertexOffsets = new int[numRings];
+          this.chainReversed = new boolean[numRings];
 
           // First pass: count total vertices and compute offsets. Sedona's WKBWriter writes
           // open rings (n unique vertices, no closing duplicate); standard WKB writes closed
@@ -149,7 +175,18 @@ public WkbS2Shape(byte[] wkb) {
             }
           }
 
-          // Eagerly compute containsOrigin from first ring
+          if (normalizePolygonRings) {
+            // Match SedonaDB's simple-features interpretation: the first ring is a shell and every
+            // subsequent ring is a hole regardless of input winding. Reverse only the S2-facing
+            // traversal, leaving the stored WKB and vertex order untouched.
+            for (int r = 0; r < numRings; r++) {
+              boolean isHole = r > 0;
+              boolean isClockwise = isClockwise(r);
+              chainReversed[r] = isHole != isClockwise;
+            }
+          }
+
+          // Compute reference containment after any virtual ring reversal has been applied.
           this.containsOriginValue = computeContainsOrigin();
           break;
         }
@@ -176,9 +213,7 @@ public void getEdge(int edgeId, MutableEdge result) {
     // Find chain
     int chainId = findChain(edgeId);
     int offset = edgeId - chainStarts[chainId];
-    int vi = vertexOffsets[chainId] + offset;
-    result.a = vertices[vi];
-    result.b = vertices[vi + 1];
+    getChainEdge(chainId, offset, result);
   }
 
   @Override
@@ -213,9 +248,14 @@ public void getChainEdge(int chainId, int offset, MutableEdge result) {
       result.b = vertices[0];
       return;
     }
-    int vi = vertexOffsets[chainId] + offset;
-    result.a = vertices[vi];
-    result.b = vertices[vi + 1];
+    int vi = vertexOffsets[chainId];
+    if (chainReversed[chainId]) {
+      result.a = vertices[vi + chainLengths[chainId] - offset];
+      result.b = vertices[vi + chainLengths[chainId] - offset - 1];
+    } else {
+      result.a = vertices[vi + offset];
+      result.b = vertices[vi + offset + 1];
+    }
   }
 
   @Override
@@ -226,7 +266,11 @@ public void getChainPosition(int edgeId, ChainPosition result) {
 
   @Override
   public S2Point getChainVertex(int chainId, int edgeOffset) {
-    return vertices[vertexOffsets[chainId] + edgeOffset];
+    if (dim == 0) {
+      return vertices[0];
+    }
+    int offset = chainReversed[chainId] ? chainLengths[chainId] - edgeOffset : edgeOffset;
+    return vertices[vertexOffsets[chainId] + offset];
   }
 
   @Override
@@ -268,40 +312,83 @@ private static S2Point[] readVertices(ByteBuffer buf, int byteOffset, int numCoo
   }
 
   /**
-   * Compute containsOrigin for polygon outer ring using direct edge-crossing test against
-   * S2.origin(). Same algorithm as S2Loop.initOriginAndBound() but without constructing an S2Loop
-   * (which builds its own internal S2ShapeIndex).
+   * Returns whether a polygon chain has negative S2 curvature (clockwise traversal). The fast path
+   * uses S2's robust turning-angle implementation through {@link S2ShapeMeasures}. Only a result
+   * within S2's documented turning-angle error bound of a half-sphere allocates an S2Loop to apply
+   * S2's hemisphere normalization convention.
    */
-  private boolean computeContainsOrigin() {
-    int start = vertexOffsets[0];
-    int numVerts = chainLengths[0]; // edges = verts - 1 for closed ring, but we use edge count
+  private boolean isClockwise(int chainId) {
+    if (chainLengths[chainId] < 3) {
+      return false;
+    }
 
-    if (numVerts < 3) return false;
+    // approxLoopArea() calls back into getChainVertex(). This method must run before assigning this
+    // chain's chainReversed entry so it measures the original WKB traversal.
+    assert !chainReversed[chainId] : "ring orientation must be measured before virtual reversal";
 
-    // Same logic as S2Loop.initOriginAndBound():
-    // 1. Guess originInside = false
-    // 2. Check if vertex(1) is inside via angle test
-    // 3. Check if contains(vertex(1)) matches — if not, flip originInside
-    S2Point v0 = vertices[start];
-    S2Point v1 = vertices[start + 1];
-    S2Point v2 = vertices[start + 2];
+    double halfSphere = 2.0 * Math.PI;
+    double loopArea = S2ShapeMeasures.approxLoopArea(this, chainId);
+    double maxError = S2.getTurningAngleMaxError(chainLengths[chainId]);
+    if (Math.abs(loopArea - halfSphere) > maxError) {
+      return loopArea > halfSphere;
+    }
 
-    boolean v1Inside =
-        !v0.equalsPoint(v1) && !v2.equalsPoint(v1) && S2Predicates.angleContainsVertex(v0, v1, v2);
+    int length = chainLengths[chainId];
+    List<S2Point> ring = new ArrayList<>(length);
+    int start = vertexOffsets[chainId];
+    for (int i = 0; i < length; i++) {
+      ring.add(vertices[start + i]);
+    }
+    return !new S2Loop(ring).isNormalized();
+  }
 
-    // Brute force contains(vertex(1)) with originInside = false
-    boolean originInside = false;
+  /**
+   * Computes S2.origin() containment. A single-ring polygon uses the same one-pass initialization
+   * as S2Loop; polygons with holes use a whole-shape reference point so every ring participates in
+   * the result.
+   */
+  private boolean computeContainsOrigin() {
+    if (numChains() == 1) {
+      return computeSingleLoopContainsOrigin(0);
+    }
+
+    ReferencePoint reference = S2ShapeUtil.getReferencePoint(this);
     S2Point origin = S2.origin();
-    S2EdgeUtil.EdgeCrosser crosser = new S2EdgeUtil.EdgeCrosser(origin, v1, v0);
-    boolean inside = originInside;
-    for (int i = 1; i <= numVerts; i++) {
-      S2Point next = vertices[start + (i % numVerts)];
-      inside ^= crosser.edgeOrVertexCrossing(next);
+    if (reference.equalsPoint(origin)) {
+      return reference.contained();
+    }
+
+    S2EdgeUtil.EdgeCrosser crosser = new S2EdgeUtil.EdgeCrosser(reference.point(), origin);
+    boolean inside = reference.contained();
+    MutableEdge edge = new MutableEdge();
+    for (int i = 0; i < numEdges(); i++) {
+      getEdge(i, edge);
+      inside ^= crosser.edgeOrVertexCrossing(edge.a, edge.b);
     }
+    return inside;
+  }
+
+  /**
+   * Computes origin containment for one polygon chain using the initialization algorithm from
+   * S2Loop, without constructing an S2Loop or first finding another reference point.
+   */
+  private boolean computeSingleLoopContainsOrigin(int chainId) {
+    int numVertices = chainLengths[chainId];
+    if (numVertices < 3) {
+      return false;
+    }
+
+    S2Point v0 = getChainVertex(chainId, 0);
+    S2Point v1 = getChainVertex(chainId, 1);
+    S2Point v2 = getChainVertex(chainId, 2);
+    boolean v1Inside =
+        !v0.equalsPoint(v1) && !v2.equalsPoint(v1) && S2Predicates.angleContainsVertex(v0, v1, v2);
 
-    if (v1Inside != inside) {
-      originInside = true;
+    S2EdgeUtil.EdgeCrosser crosser = new S2EdgeUtil.EdgeCrosser(S2.origin(), v1, v0);
+    boolean inside = false;
+    for (int i = 1; i <= numVertices; i++) {
+      inside ^= crosser.edgeOrVertexCrossing(getChainVertex(chainId, i));
     }
-    return originInside;
+    return v1Inside != inside;
   }
 }

common/src/main/java/org/apache/sedona/common/geography/Constructors.java

@@ -320,10 +320,10 @@ public static Geography geomToGeography(Geometry geom) {
     if (geom == null) {
       return null;
     }
-    // Build S2 Geography first for proper spherical normalization (e.g., deduplication),
-    // then wrap in WKBGeography for WKB-based storage.
-    Geography s2geog = geomToS2Geography(geom);
-    return WKBGeography.fromS2Geography(s2geog);
+    // Keep the input WKB as the structural source of truth. S2 is derived lazily when a spherical
+    // operation needs it; converting through S2 here would alter coordinates, collapse repeated
+    // vertices, and fail to serialize empty polygons.
+    return WKBGeography.fromJTS(geom);
   }
 
   /**