1package vnext23import (4 "bytes"5 "crypto/sha256"6 "encoding/base64"7 "encoding/binary"8 "encoding/hex"9 "encoding/json"10 "fmt"11 "io"12 "math"13 "slices"14 "strings"15)1617const (18 featureIndexName = "indexes/features.v1.json"19 featurePartitionPrefix = "data/feature-partitions/"20 featurePartitionRows = 4_09621 maxFeaturePageRows = 1_00022)2324// SpatialBounds is a closed, axis-aligned query rectangle in a world's25// declared coordinate space.26type SpatialBounds struct {27 MinX float64 `json:"minX"`28 MinY float64 `json:"minY"`29 MaxX float64 `json:"maxX"`30 MaxY float64 `json:"maxY"`31}3233type featurePartitionIndex struct {34 Version uint16 `json:"version"`35 Partitions []featurePartitionDescriptor `json:"partitions"`36}3738type featurePartitionDescriptor struct {39 FeatureSet string `json:"featureSet"`40 Key string `json:"key"`41 Bounds SpatialBounds `json:"bounds"`42 Rows int `json:"rows"`43 Features string `json:"features"`44 Relationships string `json:"relationships"`45 Provenance string `json:"provenance"`46}4748type orderedFeature struct {49 id string50 key uint6451 bounds SpatialBounds52}5354// FeaturePageRequest selects a bounded page from one semantic FeatureSet.55// After is an opaque continuation returned by the preceding page.56type FeaturePageRequest struct {57 FeatureSet string58 Bounds *SpatialBounds59 After string60 Limit int61}6263// FeaturePageResult is a demand-read page and its physical read evidence.64type FeaturePageResult struct {65 Features []Feature66 Next string67 PartitionsRead int68 BytesRead int6469}7071// FeatureSetSummary is page-zero cardinality derived from the physical index;72// reading it never opens a feature partition.73type FeatureSetSummary struct {74 FeatureSet string75 Rows int76}7778// DemandAddressedFeatures reports whether feature rows are physically79// partitioned instead of carried in the legacy monolithic tables.80func (reader *Reader) DemandAddressedFeatures() bool {81 _, held := reader.blobs[featureIndexName]82 return held83}8485// FeatureSetSummaries returns canonical per-set row counts from the physical86// partition index without materializing feature rows.87func (reader *Reader) FeatureSetSummaries() ([]FeatureSetSummary, error) {88 index, err := reader.featureIndex()89 if err != nil {90 return nil, err91 }92 if _, indexed := reader.blobs[featureIndexName]; !indexed {93 volume, err := reader.Volume()94 if err != nil {95 return nil, err96 }97 var out []FeatureSetSummary98 for _, world := range volume.Worlds {99 for _, set := range world.FeatureSets {100 out = append(out, FeatureSetSummary{FeatureSet: set.ID, Rows: len(set.Features)})101 }102 }103 slices.SortFunc(out, func(left, right FeatureSetSummary) int {104 return strings.Compare(left.FeatureSet, right.FeatureSet)105 })106 return out, nil107 }108 bySet := make(map[string]int)109 for _, partition := range index.Partitions {110 bySet[partition.FeatureSet] += partition.Rows111 }112 sets := make([]string, 0, len(bySet))113 for set := range bySet {114 sets = append(sets, set)115 }116 slices.Sort(sets)117 out := make([]FeatureSetSummary, 0, len(sets))118 for _, set := range sets {119 out = append(out, FeatureSetSummary{FeatureSet: set, Rows: bySet[set]})120 }121 return out, nil122}123124func partitionFeatureTables(volume Volume, tables map[string]Table) ([]NamedTable, Blob, error) {125 features := tables[featureTableName]126 relationships := tables[relationshipTableName]127 provenance := tables[provenanceTableName]128 featureRows := rowsByString(features, "feature.id")129 index := featurePartitionIndex{Version: 1}130 var partitions []NamedTable131 for _, world := range volume.Worlds {132 for _, set := range world.FeatureSets {133 ordered, err := orderFeatures(set.Features, world.CoordinateSpace.Extent)134 if err != nil {135 return nil, Blob{}, fmt.Errorf("partition feature set %s: %w", set.ID, err)136 }137 for start := 0; start < len(ordered); start += featurePartitionRows {138 end := min(start+featurePartitionRows, len(ordered))139 chunk := ordered[start:end]140 descriptor, named, err := makeFeaturePartition(set.ID, chunk, featureRows, features, relationships, provenance)141 if err != nil {142 return nil, Blob{}, err143 }144 index.Partitions = append(index.Partitions, descriptor)145 partitions = append(partitions, named...)146 }147 }148 }149 slices.SortFunc(index.Partitions, func(left, right featurePartitionDescriptor) int {150 if bySet := strings.Compare(left.FeatureSet, right.FeatureSet); bySet != 0 {151 return bySet152 }153 return strings.Compare(left.Key, right.Key)154 })155 data, err := json.Marshal(index)156 if err != nil {157 return nil, Blob{}, fmt.Errorf("encode feature partition index: %w", err)158 }159 return partitions, Blob{Name: featureIndexName, Data: data}, nil160}161162func orderFeatures(features []Feature, extent [4]float64) ([]orderedFeature, error) {163 ordered := make([]orderedFeature, len(features))164 for index, feature := range features {165 bounds, err := geometryBounds(feature.Geometry)166 if err != nil {167 return nil, fmt.Errorf("feature %s: %w", feature.ID, err)168 }169 ordered[index] = orderedFeature{id: feature.ID, key: spatialKey(bounds, extent), bounds: bounds}170 }171 slices.SortFunc(ordered, func(left, right orderedFeature) int {172 if left.key < right.key {173 return -1174 }175 if left.key > right.key {176 return 1177 }178 return strings.Compare(left.id, right.id)179 })180 return ordered, nil181}182183func makeFeaturePartition(184 setID string,185 ordered []orderedFeature,186 featureRows map[string]int,187 features Table,188 relationships Table,189 provenance Table,190) (featurePartitionDescriptor, []NamedTable, error) {191 ids := make(map[string]bool, len(ordered))192 rows := make([]int, len(ordered))193 bounds := ordered[0].bounds194 for index, feature := range ordered {195 row, held := featureRows[feature.id]196 if !held {197 return featurePartitionDescriptor{}, nil, fmt.Errorf("feature table omits %s", feature.id)198 }199 ids[feature.id] = true200 rows[index] = row201 bounds = unionBounds(bounds, feature.bounds)202 }203 identity := fmt.Sprintf("%s\x00%016x\x00%s\x00%016x\x00%s", setID, ordered[0].key, ordered[0].id, ordered[len(ordered)-1].key, ordered[len(ordered)-1].id)204 digest := sha256.Sum256([]byte(identity))205 token := hex.EncodeToString(digest[:12])206 base := featurePartitionPrefix + token207 descriptor := featurePartitionDescriptor{208 FeatureSet: setID, Key: fmt.Sprintf("%016x/%s", ordered[0].key, ordered[0].id), Bounds: bounds, Rows: len(rows),209 Features: base + ".features.pack", Relationships: base + ".relationships.pack", Provenance: base + ".provenance.pack",210 }211 named := []NamedTable{212 {Name: descriptor.Features, Table: selectRows(features, rows)},213 {Name: descriptor.Relationships, Table: selectRowsForIDs(relationships, "relationship.feature", ids)},214 {Name: descriptor.Provenance, Table: selectRowsForIDs(provenance, "provenance.feature", ids)},215 }216 return descriptor, named, nil217}218219func rowsByString(table Table, field string) map[string]int {220 rows := make(map[string]int, table.Rows)221 for row := 0; row < table.Rows; row++ {222 rows[requiredString(table, row, field)] = row223 }224 return rows225}226227func selectRowsForIDs(table Table, field string, ids map[string]bool) Table {228 rows := make([]int, 0)229 for row := 0; row < table.Rows; row++ {230 if ids[requiredString(table, row, field)] {231 rows = append(rows, row)232 }233 }234 slices.SortFunc(rows, func(left, right int) int {235 return bytes.Compare(stableRowKey(table, left), stableRowKey(table, right))236 })237 return selectRows(table, rows)238}239240func stableRowKey(table Table, row int) []byte {241 values := make([]Value, len(table.Columns))242 for index, column := range table.Columns {243 values[index] = column.Values[row]244 }245 encoded, _ := json.Marshal(values)246 return encoded247}248249func selectRows(table Table, rows []int) Table {250 out := Table{TypeID: table.TypeID, Rows: len(rows), Columns: make([]Column, len(table.Columns))}251 for columnIndex, column := range table.Columns {252 values := make([]Value, len(rows))253 for index, row := range rows {254 values[index] = column.Values[row]255 }256 out.Columns[columnIndex] = Column{Field: column.Field, Values: values}257 }258 return canonicalTable(out)259}260261func geometryBounds(geometry Geometry) (SpatialBounds, error) {262 bounds := SpatialBounds{MinX: math.Inf(1), MinY: math.Inf(1), MaxX: math.Inf(-1), MaxY: math.Inf(-1)}263 for _, part := range geometry.Parts {264 for _, ring := range part.Rings {265 for _, position := range ring {266 bounds.MinX = math.Min(bounds.MinX, position[0])267 bounds.MinY = math.Min(bounds.MinY, position[1])268 bounds.MaxX = math.Max(bounds.MaxX, position[0])269 bounds.MaxY = math.Max(bounds.MaxY, position[1])270 }271 }272 }273 if math.IsInf(bounds.MinX, 0) {274 return SpatialBounds{}, fmt.Errorf("geometry has no positions")275 }276 return bounds, nil277}278279func spatialKey(bounds SpatialBounds, extent [4]float64) uint64 {280 x := quantize((bounds.MinX+bounds.MaxX)/2, extent[0], extent[2])281 y := quantize((bounds.MinY+bounds.MaxY)/2, extent[1], extent[3])282 return morton32(x, y)283}284285func quantize(value, minimum, maximum float64) uint32 {286 if maximum <= minimum || value <= minimum {287 return 0288 }289 if value >= maximum {290 return math.MaxUint32291 }292 return uint32((value - minimum) / (maximum - minimum) * float64(math.MaxUint32))293}294295func morton32(x, y uint32) uint64 {296 var out uint64297 for bit := range 32 {298 out |= uint64((x>>bit)&1) << (bit * 2)299 out |= uint64((y>>bit)&1) << (bit*2 + 1)300 }301 return out302}303304func unionBounds(left, right SpatialBounds) SpatialBounds {305 return SpatialBounds{306 MinX: math.Min(left.MinX, right.MinX), MinY: math.Min(left.MinY, right.MinY),307 MaxX: math.Max(left.MaxX, right.MaxX), MaxY: math.Max(left.MaxY, right.MaxY),308 }309}310311func intersects(left, right SpatialBounds) bool {312 return left.MinX <= right.MaxX && left.MaxX >= right.MinX && left.MinY <= right.MaxY && left.MaxY >= right.MinY313}314315func (reader *Reader) featureIndex() (featurePartitionIndex, error) {316 reference, held := reader.blobs[featureIndexName]317 if !held {318 return featurePartitionIndex{}, nil319 }320 data, err := readReferencedEntry(reader.entries[featureIndexName], reference.Length, reference.Hash, reader.limits.MaxBlobBytes)321 if err != nil {322 return featurePartitionIndex{}, err323 }324 var index featurePartitionIndex325 decoder := json.NewDecoder(bytes.NewReader(data))326 decoder.DisallowUnknownFields()327 if err := decoder.Decode(&index); err != nil || index.Version != 1 {328 return featurePartitionIndex{}, fmt.Errorf("decode feature partition index")329 }330 if err := decoder.Decode(&struct{}{}); err != io.EOF {331 return featurePartitionIndex{}, fmt.Errorf("feature partition index carries trailing data")332 }333 canonical, err := json.Marshal(index)334 if err != nil || !bytes.Equal(canonical, data) {335 return featurePartitionIndex{}, fmt.Errorf("feature partition index is not canonical")336 }337 if err := reader.validateFeatureIndex(index); err != nil {338 return featurePartitionIndex{}, err339 }340 return index, nil341}342343func (reader *Reader) validateFeatureIndex(index featurePartitionIndex) error {344 seen := make(map[string]bool, len(index.Partitions)*3)345 previous := ""346 for _, partition := range index.Partitions {347 key := partition.FeatureSet + "\x00" + partition.Key348 finite := !math.IsNaN(partition.Bounds.MinX) && !math.IsNaN(partition.Bounds.MinY) &&349 !math.IsNaN(partition.Bounds.MaxX) && !math.IsNaN(partition.Bounds.MaxY) &&350 !math.IsInf(partition.Bounds.MinX, 0) && !math.IsInf(partition.Bounds.MinY, 0) &&351 !math.IsInf(partition.Bounds.MaxX, 0) && !math.IsInf(partition.Bounds.MaxY, 0)352 if partition.FeatureSet == "" || partition.Key == "" || key <= previous || partition.Rows <= 0 ||353 partition.Rows > featurePartitionRows || !finite || partition.Bounds.MinX > partition.Bounds.MaxX ||354 partition.Bounds.MinY > partition.Bounds.MaxY {355 return fmt.Errorf("feature partition index is not canonical")356 }357 for _, expected := range []struct {358 name string359 typeID ID360 }{361 {partition.Features, CoreID("type.feature")},362 {partition.Relationships, CoreID("type.relationship")},363 {partition.Provenance, CoreID("type.provenance")},364 } {365 reference, held := reader.tables[expected.name]366 if !held || seen[expected.name] || !strings.HasPrefix(expected.name, featurePartitionPrefix) || reference.RootType != expected.typeID {367 return fmt.Errorf("feature partition %s has an invalid table role", partition.Key)368 }369 seen[expected.name] = true370 }371 previous = key372 }373 for name := range reader.tables {374 if strings.HasPrefix(name, featurePartitionPrefix) && !seen[name] {375 return fmt.Errorf("feature partition table %s is not indexed", name)376 }377 }378 return nil379}380381func (reader *Reader) partitionedTable(name string) (Table, bool, error) {382 index, err := reader.featureIndex()383 if err != nil || len(index.Partitions) == 0 {384 return Table{}, false, err385 }386 var names []string387 for _, partition := range index.Partitions {388 switch name {389 case featureTableName:390 names = append(names, partition.Features)391 case relationshipTableName:392 names = append(names, partition.Relationships)393 case provenanceTableName:394 names = append(names, partition.Provenance)395 default:396 return Table{}, false, nil397 }398 }399 var combined Table400 for _, physical := range names {401 table, err := reader.Table(physical)402 if err != nil {403 return Table{}, true, err404 }405 combined, err = appendTable(combined, table)406 if err != nil {407 return Table{}, true, err408 }409 }410 return combined, true, nil411}412413func appendTable(left, right Table) (Table, error) {414 if left.TypeID.isZero() {415 return right, nil416 }417 if left.TypeID != right.TypeID || len(left.Columns) != len(right.Columns) {418 return Table{}, fmt.Errorf("partition table shapes disagree")419 }420 out := Table{TypeID: left.TypeID, Rows: left.Rows + right.Rows, Columns: make([]Column, len(left.Columns))}421 for index := range left.Columns {422 if left.Columns[index].Field.ID != right.Columns[index].Field.ID {423 return Table{}, fmt.Errorf("partition columns disagree")424 }425 values := append([]Value(nil), left.Columns[index].Values...)426 values = append(values, right.Columns[index].Values...)427 out.Columns[index] = Column{Field: left.Columns[index].Field, Values: values}428 }429 return out, nil430}431432// FeaturePage reads only intersecting partitions until the requested page is433// full. It never calls Volume and its physical counters make that observable.434func (reader *Reader) FeaturePage(request FeaturePageRequest) (FeaturePageResult, error) {435 if request.FeatureSet == "" || request.Limit < 0 || request.Limit > maxFeaturePageRows {436 return FeaturePageResult{}, fmt.Errorf("invalid feature page request")437 }438 if request.Limit == 0 {439 request.Limit = 100440 }441 after, err := decodeContinuation(request)442 if err != nil {443 return FeaturePageResult{}, err444 }445 index, err := reader.featureIndex()446 if err != nil {447 return FeaturePageResult{}, err448 }449 if _, indexed := reader.blobs[featureIndexName]; !indexed {450 return reader.legacyFeaturePage(request, after)451 }452 result := FeaturePageResult{}453 passedAfter := after == ""454 for _, partition := range index.Partitions {455 if partition.FeatureSet != request.FeatureSet || request.Bounds != nil && !intersects(partition.Bounds, *request.Bounds) {456 continue457 }458 features, bytesRead, err := reader.readFeaturePartition(partition)459 if err != nil {460 return FeaturePageResult{}, err461 }462 result.PartitionsRead++463 result.BytesRead += bytesRead464 for _, feature := range features {465 if !passedAfter {466 passedAfter = feature.ID == after467 continue468 }469 if request.Bounds != nil {470 bounds, err := geometryBounds(feature.Geometry)471 if err != nil || !intersects(bounds, *request.Bounds) {472 continue473 }474 }475 result.Features = append(result.Features, feature)476 if len(result.Features) == request.Limit {477 result.Next = encodeContinuation(request, feature.ID)478 return result, nil479 }480 }481 }482 if !passedAfter {483 return FeaturePageResult{}, fmt.Errorf("feature continuation is not present")484 }485 return result, nil486}487488func (reader *Reader) legacyFeaturePage(request FeaturePageRequest, after string) (FeaturePageResult, error) {489 volume, err := reader.Volume()490 if err != nil {491 return FeaturePageResult{}, err492 }493 result := FeaturePageResult{PartitionsRead: 1}494 for _, name := range []string{featureTableName, relationshipTableName, provenanceTableName} {495 result.BytesRead += reader.tables[name].Length496 }497 passedAfter := after == ""498 foundSet := false499 for _, world := range volume.Worlds {500 for _, set := range world.FeatureSets {501 if set.ID != request.FeatureSet {502 continue503 }504 foundSet = true505 for _, feature := range set.Features {506 if !passedAfter {507 passedAfter = feature.ID == after508 continue509 }510 if request.Bounds != nil {511 bounds, err := geometryBounds(feature.Geometry)512 if err != nil || !intersects(bounds, *request.Bounds) {513 continue514 }515 }516 result.Features = append(result.Features, feature)517 if len(result.Features) == request.Limit {518 result.Next = encodeContinuation(request, feature.ID)519 return result, nil520 }521 }522 }523 }524 if !foundSet {525 return FeaturePageResult{}, fmt.Errorf("feature set is not present")526 }527 if !passedAfter {528 return FeaturePageResult{}, fmt.Errorf("feature continuation is not present")529 }530 return result, nil531}532533func (reader *Reader) readFeaturePartition(partition featurePartitionDescriptor) ([]Feature, int64, error) {534 features, err := reader.Table(partition.Features)535 if err != nil {536 return nil, 0, err537 }538 relationships, err := reader.Table(partition.Relationships)539 if err != nil {540 return nil, 0, err541 }542 provenance, err := reader.Table(partition.Provenance)543 if err != nil {544 return nil, 0, err545 }546 out := make([]Feature, features.Rows)547 byID := make(map[string]int, features.Rows)548 core, _ := StandardSchema().Type(CoreID("type.feature"))549 for row := 0; row < features.Rows; row++ {550 geometry, err := decodeGeometry(requiredBytes(features, row, "feature.geometry"))551 if err != nil {552 return nil, 0, err553 }554 feature := Feature{555 ID: requiredString(features, row, "feature.id"), Title: requiredString(features, row, "feature.title"),556 Subtitle: optionalStringAt(features, row, "feature.subtitle"), Description: optionalStringAt(features, row, "feature.description"),557 Shard: requiredInt64(features, row, "feature.shard"), Geometry: geometry,558 }559 if center := optionalBytesAt(features, row, "feature.center"); center != nil {560 position, err := decodePosition(center)561 if err != nil {562 return nil, 0, err563 }564 feature.Center = &position565 }566 for _, column := range features.Columns {567 if _, standard := core.Field(column.Field.ID); !standard && column.Values[row].Kind != KindInvalid {568 feature.Properties = append(feature.Properties, Property{FieldID: column.Field.ID, Field: column.Field, Value: column.Values[row]})569 }570 }571 out[row] = feature572 byID[feature.ID] = row573 }574 for row := 0; row < relationships.Rows; row++ {575 if at, held := byID[requiredString(relationships, row, "relationship.feature")]; held {576 decodeRelationship(&out[at], relationships, row)577 }578 }579 for row := 0; row < provenance.Rows; row++ {580 if at, held := byID[requiredString(provenance, row, "provenance.feature")]; held {581 decodeProvenance(&out[at], provenance, row)582 }583 }584 bytesRead := reader.tables[partition.Features].Length + reader.tables[partition.Relationships].Length + reader.tables[partition.Provenance].Length585 return out, bytesRead, nil586}587588func continuationFingerprint(request FeaturePageRequest) [8]byte {589 hash := sha256.New()590 _, _ = hash.Write([]byte(request.FeatureSet))591 if request.Bounds != nil {592 for _, value := range []float64{request.Bounds.MinX, request.Bounds.MinY, request.Bounds.MaxX, request.Bounds.MaxY} {593 var encoded [8]byte594 binary.LittleEndian.PutUint64(encoded[:], math.Float64bits(value))595 _, _ = hash.Write(encoded[:])596 }597 }598 var out [8]byte599 copy(out[:], hash.Sum(nil))600 return out601}602603func encodeContinuation(request FeaturePageRequest, featureID string) string {604 fingerprint := continuationFingerprint(request)605 data := append(fingerprint[:], featureID...)606 return base64.RawURLEncoding.EncodeToString(data)607}608609func decodeContinuation(request FeaturePageRequest) (string, error) {610 if request.After == "" {611 return "", nil612 }613 data, err := base64.RawURLEncoding.DecodeString(request.After)614 if err != nil || len(data) <= 8 {615 return "", fmt.Errorf("feature continuation is invalid")616 }617 fingerprint := continuationFingerprint(request)618 if !slices.Equal(data[:8], fingerprint[:]) {619 return "", fmt.Errorf("feature continuation belongs to another query")620 }621 return string(data[8:]), nil622}