C-Loftus · January 10, 2026 17:46 · C-Loftus · Jan 10, 2026
diff --git a/main.go b/main.go
 package main

 import (
 	"database/sql"
 	"errors"
 	"fmt"
 	"sync"

 	_ "github.com/duckdb/duckdb-go/v2"
 	geom "github.com/peterstace/simplefeatures/geom"
 )

 type S3FlatgeobufMainstemService struct {
 	duckdb *sql.DB

 	mainstemFlatgeobufURI string
 }

 type MainstemQueryResponse struct {
 	foundAssociatedMainstem bool
 	mainstemURI             string
 }

 func NewS3FlatgeobufMainstemService(mainstemFlatgeobufURI string) (S3FlatgeobufMainstemService, error) {
 	db, err := sql.Open("duckdb", "")
 	if err != nil {
 		return S3FlatgeobufMainstemService{}, err
 	}
 	_, err = db.Exec("INSTALL spatial; LOAD spatial;")
 	if err != nil {
 		return S3FlatgeobufMainstemService{}, err
 	}
 	return S3FlatgeobufMainstemService{duckdb: db, mainstemFlatgeobufURI: mainstemFlatgeobufURI}, nil
 }

 func (s S3FlatgeobufMainstemService) GetMainstemForWkt(wkt string) (MainstemQueryResponse, error) {
 	geometry, err := geom.UnmarshalWKT(wkt)
 	if err != nil {
 		return MainstemQueryResponse{}, err
 	}
 	point := geometry.Centroid()
 	coordinates, isNonEmpty := point.Coordinates()
 	if !isNonEmpty {
 		return MainstemQueryResponse{}, fmt.Errorf("got an empty centroid result for WKT: %s", wkt)
 	}

 	// flatgeobuf requires opening with a bbox in duckdb
 	// in order to subset the data; by using the same
 	// value for min and max we get a specific point
 	// and a guarantee of no overlaps
 	mainstemSQL := `
    SELECT geoconnex_url
 		FROM ST_Read(
 			?,
 			spatial_filter_box = ST_MakeBox2D(
 				ST_Point(?, ?),
 				ST_Point(?, ?)
 			)
 		)
 	`
 	result := s.duckdb.QueryRow(mainstemSQL, s.mainstemFlatgeobufURI, coordinates.X, coordinates.Y, coordinates.X, coordinates.Y)
 	if result.Err() != nil {
 		return MainstemQueryResponse{}, fmt.Errorf("mainstem query failed for %s: %w", wkt, result.Err())
 	}
 	var mainstemURI sql.NullString
 	if err := result.Scan(&mainstemURI); err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return MainstemQueryResponse{foundAssociatedMainstem: false, mainstemURI: ""}, nil
 		}
 		return MainstemQueryResponse{}, fmt.Errorf("failed to get sql result for query at %s, %v", wkt, err)
 	}
 	if mainstemURI.Valid && mainstemURI.String != "" {
 		return MainstemQueryResponse{
 			foundAssociatedMainstem: true,
 			mainstemURI:             mainstemURI.String,
 		}, nil
 	}
 	return MainstemQueryResponse{
 		foundAssociatedMainstem: false,
 		mainstemURI:             "",
 	}, nil
 }

 func main() {
 	const mainstemFlatgeobufURI = "gcs://national-hydrologic-geospatial-fabric-reference-hydrofabric/reference_catchments_and_flowlines.fgb"

 	wg := sync.WaitGroup{}
 	service, err := NewS3FlatgeobufMainstemService(mainstemFlatgeobufURI)
 	if err != nil {
 		panic(err)
 	}

 	// create a list of test WKTs
 	testWKTs := []string{}
 	// 10 is arbitrary; it has a chance to fail regardless of the amount of concurrency as long as there is some concurrency
 	// i have seen failures with just 3 concurrent queries to the flatgeobuf
 	for range 10 {
 		testWKTs = append(testWKTs, []string{
 			"POINT (-89.398 43.0731)",
 			"POINT (-90.0 44.0)",
 			"POINT (-88.5 42.5)",
 		}...)
 	}

 	fmt.Print("Starting queries")
 	for _, wkt := range testWKTs {
 		wg.Add(1)
 		go func(wkt string) {
 			defer wg.Done()
 			_, err := service.GetMainstemForWkt(wkt)
 			if err != nil {
 				fmt.Printf("Error querying mainstem for WKT %s: %v\n", wkt, err)
 				return
 			}
 		}(wkt)
 	}

 	wg.Wait()
 	fmt.Println("Completed all flatgeobuf queries.")
 }
	package main

	import (
	"database/sql"
	"errors"
	"fmt"
	"sync"

	_ "github.com/duckdb/duckdb-go/v2"
	geom "github.com/peterstace/simplefeatures/geom"
	)

	type S3FlatgeobufMainstemService struct {
	duckdb *sql.DB

	mainstemFlatgeobufURI string
	}

	type MainstemQueryResponse struct {
	foundAssociatedMainstem bool
	mainstemURI string
	}

	func NewS3FlatgeobufMainstemService(mainstemFlatgeobufURI string) (S3FlatgeobufMainstemService, error) {
	db, err := sql.Open("duckdb", "")
	if err != nil {
	return S3FlatgeobufMainstemService{}, err
	}
	_, err = db.Exec("INSTALL spatial; LOAD spatial;")
	if err != nil {
	return S3FlatgeobufMainstemService{}, err
	}
	return S3FlatgeobufMainstemService{duckdb: db, mainstemFlatgeobufURI: mainstemFlatgeobufURI}, nil
	}

	func (s S3FlatgeobufMainstemService) GetMainstemForWkt(wkt string) (MainstemQueryResponse, error) {
	geometry, err := geom.UnmarshalWKT(wkt)
	if err != nil {
	return MainstemQueryResponse{}, err
	}
	point := geometry.Centroid()
	coordinates, isNonEmpty := point.Coordinates()
	if !isNonEmpty {
	return MainstemQueryResponse{}, fmt.Errorf("got an empty centroid result for WKT: %s", wkt)
	}

	// flatgeobuf requires opening with a bbox in duckdb
	// in order to subset the data; by using the same
	// value for min and max we get a specific point
	// and a guarantee of no overlaps
	mainstemSQL := `
	SELECT geoconnex_url
	FROM ST_Read(
	?,
	spatial_filter_box = ST_MakeBox2D(
	ST_Point(?, ?),
	ST_Point(?, ?)
	)
	)
	`
	result := s.duckdb.QueryRow(mainstemSQL, s.mainstemFlatgeobufURI, coordinates.X, coordinates.Y, coordinates.X, coordinates.Y)
	if result.Err() != nil {
	return MainstemQueryResponse{}, fmt.Errorf("mainstem query failed for %s: %w", wkt, result.Err())
	}
	var mainstemURI sql.NullString
	if err := result.Scan(&mainstemURI); err != nil {
	if errors.Is(err, sql.ErrNoRows) {
	return MainstemQueryResponse{foundAssociatedMainstem: false, mainstemURI: ""}, nil
	}
	return MainstemQueryResponse{}, fmt.Errorf("failed to get sql result for query at %s, %v", wkt, err)
	}
	if mainstemURI.Valid && mainstemURI.String != "" {
	return MainstemQueryResponse{
	foundAssociatedMainstem: true,
	mainstemURI: mainstemURI.String,
	}, nil
	}
	return MainstemQueryResponse{
	foundAssociatedMainstem: false,
	mainstemURI: "",
	}, nil
	}

	func main() {
	const mainstemFlatgeobufURI = "gcs://national-hydrologic-geospatial-fabric-reference-hydrofabric/reference_catchments_and_flowlines.fgb"

	wg := sync.WaitGroup{}
	service, err := NewS3FlatgeobufMainstemService(mainstemFlatgeobufURI)
	if err != nil {
	panic(err)
	}

	// create a list of test WKTs
	testWKTs := []string{}
	// 10 is arbitrary; it has a chance to fail regardless of the amount of concurrency as long as there is some concurrency
	// i have seen failures with just 3 concurrent queries to the flatgeobuf
	for range 10 {
	testWKTs = append(testWKTs, []string{
	"POINT (-89.398 43.0731)",
	"POINT (-90.0 44.0)",
	"POINT (-88.5 42.5)",
	}...)
	}

	fmt.Print("Starting queries")
	for _, wkt := range testWKTs {
	wg.Add(1)
	go func(wkt string) {
	defer wg.Done()
	_, err := service.GetMainstemForWkt(wkt)
	if err != nil {
	fmt.Printf("Error querying mainstem for WKT %s: %v\n", wkt, err)
	return
	}
	}(wkt)
	}

	wg.Wait()
	fmt.Println("Completed all flatgeobuf queries.")
	}
No results found