Merge #50457 #50560

50457: geo: coerce invalid geography coordinates into geometry r=otan a=Arun4rangan

This commit resolves: #50219

Previously geometry did not convert out of range lat-lngs into
correct range. This commit mimics the behavior found in PostGIS.
It converts out of range lat-lngs into correct range.

Release note (sql change): Geometry now coerce invalid
geography coordinates into correct geometry.

50560: sql: add support for aggregations and values in the new factory r=yuzefovich a=yuzefovich

**sql: remove some unnecessary things around aggregations**

Release note: None

**sql: add support for aggregations in the new factory**

This commit implements `ConstructGroupBy` and `ConstructScalarGroupBy`
methods in the new factory by populating the specs upfront and reusing
DistSQLPlanner to do the actual planning.

Fixes: #50290.

Release note: None

**sql: minor cleanup of planNodeToRowSource**

This commit cleans up `planNodeToRowSource` to be properly closed
(similar to other processors).

Release note: None

**sql: implement ConstructValues in the new factory**

This commit adds implementation of `ConstructValues` in the new factory.
In some cases, a valuesNode is the only way to "construct values" - when
the node must be wrapped (see createPhysPlanForValuesNode for more
details). In other cases, a valuesNode is used to construct the values
processor spec. The latter usage is refactored to avoid valuesNode
creation.

This decision also prompts us to add a separate "side" list of
`planNode`s that are part of the physical plan and need to be closed
when the whole plan is closed. This is done by introducing a utility
wrapper around `PhysicalPlan`. This approach was chosen after
considering an alternative in which `planNodeToRowSource` adapter would
be the one closing the `planNode` it is wrapping because
`planNode.Close` contract currently prohibits the execution from closing
any of the `planNode`s, and changined that would be quite invasive at
this point. We might reevaluate this decision later, once we've made
more progress on the distsql spec work.

Addresses: #47473.

Release note: None

**logictest: add spec-planning configs to the default ones**

We have now implemented noticeable chunk of methods in the new factory,
so I think it makes to run all logic tests with the spec-planning
configs by default.

The only logic test that is currently skipped is `interleaved_join`
because the new factory doesn't plan interleaved joins yet.

Release note: None

**sql: enhance unsupported error message in the new factory**

Release note: None

**sql: support simple projection on top of planNode in the new factory**

The new factory still constructs some of the `planNode`s (for example,
explain variants), and we should be able to handle simple projections on
top of those. This is handled by reusing the logic from the old factory.
The issue was hidden by the fallback to the old factory since EXPLAIN
statements don't have "SELECT" statement tag.

Release note: None

**sql: implement ConstructLimit in the new factory**

This commit additionally removes the requirement that `limitExpr` and
`offsetExpr` must be distributable in order for the whole plan to be
distributable in the old path because those expressions are evaluated
during the physical planning, locally.

Release note: None

Co-authored-by: Arun Ranganathan <arun.ranga@hotmail.ca>
Co-authored-by: Yahor Yuzefovich <yahor@cockroachlabs.com>

pkg/geo/geo.go

@@ -617,8 +617,38 @@ func S2RegionsFromGeom(geomRepr geom.T, emptyBehavior EmptyBehavior) ([]s2.Regio
 // Common
 //
 
+// normalizeLngLat normalizes geographical coordinates into a valid range.
+func normalizeLngLat(lng float64, lat float64) (float64, float64) {
+	if lat > 90 || lat < -90 {
+		lat = NormalizeLatitudeDegrees(lat)
+	}
+	if lng > 180 || lng < -180 {
+		lng = NormalizeLongitudeDegrees(lng)
+	}
+	return lng, lat
+}
+
+// normalizeGeographyGeomT limits geography coordinates to spherical coordinates
+// by converting geom.T cordinates inplace
+func normalizeGeographyGeomT(t geom.T) {
+	switch repr := t.(type) {
+	case *geom.GeometryCollection:
+		for _, geom := range repr.Geoms() {
+			normalizeGeographyGeomT(geom)
+		}
+	default:
+		coords := repr.FlatCoords()
+		for i := 0; i < len(coords); i += repr.Stride() {
+			coords[i], coords[i+1] = normalizeLngLat(coords[i], coords[i+1])
+		}
+	}
+}
+
 // spatialObjectFromGeomT creates a geopb.SpatialObject from a geom.T.
 func spatialObjectFromGeomT(t geom.T, soType geopb.SpatialObjectType) (geopb.SpatialObject, error) {
+	if soType == geopb.SpatialObjectType_GeographyType {
+		normalizeGeographyGeomT(t)
+	}
 	ret, err := ewkb.Marshal(t, DefaultEWKBEncodingFormat)
 	if err != nil {
 		return geopb.SpatialObject{}, err

pkg/geo/geo_test.go

@@ -120,6 +120,78 @@ func TestGeospatialTypeFitsColumnMetadata(t *testing.T) {
 	}
 }
 
+func TestMakeValidGeographyGeom(t *testing.T) {
+	var (
+		invalidGeomPoint              = geom.NewPointFlat(geom.XY, []float64{200.0, 199.0})
+		invalidGeomLineString         = geom.NewLineStringFlat(geom.XY, []float64{90.0, 90.0, 180.0, 180.0})
+		invalidGeomPolygon            = geom.NewPolygonFlat(geom.XY, []float64{360.0, 360.0, 450.0, 450.0, 540.0, 540.0, 630.0, 630.0}, []int{8})
+		invalidGeomMultiPoint         = geom.NewMultiPointFlat(geom.XY, []float64{-90.0, -90.0, -180.0, -180.0})
+		invalidGeomMultiLineString    = geom.NewMultiLineStringFlat(geom.XY, []float64{-270.0, -270.0, -360.0, -360.0, -450.0, -450.0, -540.0, -540.0}, []int{4, 8})
+		invalidGeomMultiPolygon       = geom.NewMultiPolygon(geom.XY)
+		invalidGeomGeometryCollection = geom.NewGeometryCollection()
+	)
+	invalidGeomGeometryCollection.MustPush(geom.NewPointFlat(geom.XY, []float64{200.0, 199.0}))
+	invalidGeomGeometryCollection.MustPush(geom.NewLineStringFlat(geom.XY, []float64{90.0, 90.0, 180.0, 180.0}))
+	var (
+		validGeomPoint              = geom.NewPointFlat(geom.XY, []float64{-160.0, -19.0})
+		validGeomLineString         = geom.NewLineStringFlat(geom.XY, []float64{90.0, 90.0, 180.0, 0.0})
+		validGeomPolygon            = geom.NewPolygonFlat(geom.XY, []float64{0.0, 0.0, 90.0, 90.0, -180.0, 0.0, -90.0, -90.0}, []int{8})
+		validGeomMultiPoint         = geom.NewMultiPointFlat(geom.XY, []float64{-90.0, -90.0, -180.0, 0.0})
+		validGeomMultiLineString    = geom.NewMultiLineStringFlat(geom.XY, []float64{90.0, 90.0, 0.0, 0.0, -90.0, -90.0, 180.0, 0.0}, []int{4, 8})
+		validGeomMultiPolygon       = geom.NewMultiPolygon(geom.XY)
+		validGeomGeometryCollection = geom.NewGeometryCollection()
+	)
+	validGeomGeometryCollection.MustPush(validGeomPoint)
+	validGeomGeometryCollection.MustPush(validGeomLineString)
+	testCases := []struct {
+		desc string
+		g    geom.T
+		ret  geom.T
+	}{
+		{
+			"Point",
+			invalidGeomPoint,
+			validGeomPoint,
+		},
+		{
+			"linestring",
+			invalidGeomLineString,
+			validGeomLineString,
+		},
+		{
+			"polygon",
+			invalidGeomPolygon,
+			validGeomPolygon,
+		},
+		{
+			"multipoint",
+			invalidGeomMultiPoint,
+			validGeomMultiPoint,
+		},
+		{
+			"multilinestring",
+			invalidGeomMultiLineString,
+			validGeomMultiLineString,
+		},
+		{
+			"multipolygon",
+			invalidGeomMultiPolygon,
+			validGeomMultiPolygon,
+		},
+		{
+			"geometrycollection",
+			invalidGeomGeometryCollection,
+			validGeomGeometryCollection,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.desc, func(t *testing.T) {
+			normalizeGeographyGeomT(tc.g)
+			require.Equal(t, tc.ret, tc.g)
+		})
+	}
+}
+
 func TestSpatialObjectFromGeomT(t *testing.T) {
 	testCases := []struct {
 		desc   string