gobi

package module
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 22, 2026 License: MIT Imports: 18 Imported by: 0

README

gobi

gobi is a geospatial dataframe library for Go, built on top of Apache Arrow. Think of it as a GeoPandas-shaped API with Polars-shaped internals: columnar, chunk-slice fast paths, and built around a strongly-typed schema.

Status: early. The API is settled enough to build a small pipeline on; semver stability begins with the first tagged release. GeoParquet v1.1 output has been verified against GeoPandas v1.1.1 and QGIS v4.0.2.

Highlights

  • Arrow-native. A Frame is a set of Arrow Columns. Head, Tail, and row selection are zero-copy where possible.
  • Full 2D + optional XYZ geometry. Point, LineString, Polygon (with holes), MultiPoint, MultiLineString, MultiPolygon, GeometryCollection. WKB and WKT round-trip per OGC/ISO SFA 1.2 (type codes 1..7 for 2D, 1001..1007 for XYZ).
  • Real spatial operations. Area, length, centroid (every geometry type), convex hull, containment, Simplify (Douglas-Peucker), Buffer with rounded joins and caps, EstimateUTMCRS on every type.
  • Geometry constructors from columns. PointsFromXY(x, y, crs) and PointsFromXYZ(x, y, z, crs) build a WKB geometry Series directly from numeric coordinate columns — modeled on geopandas.points_from_xy. Mixed numeric types (Float64, Float32, Int64, Int32) auto-promote; nulls in either input yield a null geometry.
  • Reprojection engine. WGS84 ↔ Web Mercator ↔ all 120 UTM zones, using the ellipsoidal Redfearn/Snyder formulas. Sub-cm round-trip accuracy verified against reference cities worldwide.
  • Spatial index and join. Static Sort-Tile-Recursive R-tree with bounding-box and k-nearest queries. Frame.SJoin(right, ..., pred) with SPIntersects / SPContains / SPWithin predicates, multi-threaded across left rows, tunable via Workers(n).
  • DataFrame ops. Filter, Take, Head, Tail, SortBy (multi-key stable, nulls-last), WithColumn, DropColumn, GroupBy(...).Agg(count/sum/mean/min/max), Join (inner / left / right / full / semi / anti with coalesced keys), Explode. Series arithmetic (Add/Sub/Mul/Div + scalar), comparisons, aggregations — all with single-chunk bulk fast paths.
  • User-defined aggregations. type Aggregator interface { ... } plugs directly into GroupBy.Agg alongside the built-ins — mode, percentile, weighted mean, H3-of-centroid, whatever you need, without forking the package.
  • Expression IR. gobi.Col("price").Mul(gobi.Lit(1.08)).Gt(gobi.Lit(100)) builds a data tree, not a chain of already-executed calls. Frame.FilterExpr and Frame.WithColumnExpr evaluate it; a Custom(node ExprNode) escape hatch lets sibling packages (H3, hashes, ML inference) plug in their own expression types alongside the built-ins.
  • LazyFrame + rule-based optimizer. df.Lazy() and parquetio.ScanFile(path) build plan trees that don't execute until .Collect(). Nine rewrite rules run to a fixed point: constant folding, dead-filter removal, adjacent-filter combining, push-filter-below-project, push-filter-below-sort, column projection pushdown, predicate pushdown (into row-group stats), and cascade-empty (short-circuits Lit(false)-derived subtrees). Projection pushdown routes into parquetio.ReadOptions.Columns — 2.4× faster reads on partial-column queries, matching the eager baseline. Optimizer overhead is ~8 µs on a five-node plan; always-on.
  • Parallel streaming executor. LazyFrame.Collect() compiles the optimized plan to a tree of ExecOperators that pull one record batch at a time — bounded memory regardless of source size. Filter, Project, WithColumn, Drop, Limit, ScanFrame, and ScanFile all stream natively. Aggregate (built-in Kinds) and hash-join (Inner/Left/Semi/Anti) run as native streaming operators too — no materialization step. Parquet scan parallelizes across row-groups; the streaming hash aggregate partitions rows across workers by key hash. Both scale to GOMAXPROCS out of the box. LazyFrame.ExplainPhysical() prints what strategy each node compiles to (worker counts included).
  • Datetime + timezone-aware ops. Timestamp[ns] columns with optional IANA tz label. Component extractors, AddDuration / DiffDuration, comparisons, sub-day + calendar truncation. ResampleEvery(timeCol, interval) for downsampling and RollingBy(timeCol, period) for trailing time windows; plus fixed-size Series.RollingSum / Mean / Min / Max / Count.
  • Multi-format I/O. Every format has ReadFile / WriteFile at Frame level plus a ScanFile LazyFrame entry point where the underlying source supports it. Formats: CSV (with .gz / .zst / .bz2 auto-detect), Parquet with proper GeoParquet 1.1 metadata (snappy / gzip / brotli / zstd / lz4), full RFC 7946 GeoJSON (every geometry type + XYZ, .geojsonl streaming), OGC GeoPackage 1.3 (SQLite, RTree spatial index, spec-compliant metadata), PostgreSQL / PostGIS (via pgx/v5, native CopyFrom bulk load), KML read/write, and Shapefile read/write (.shp + .shx + .dbf + optional .prj).
  • Streaming readers. csvio.ReadFileChunksFunc and parquetio.ReadFileChunksFunc yield one Frame per record batch (~64k rows), releasing arrow buffers after each callback. Peak memory is bounded regardless of source-file size — good for ETL over multi-GB inputs.
  • Column projection. parquetio.ReadOptions{Columns: ...} skips fetch, decompress, and arrow materialization for the columns you don't need. Composes with streaming.
  • Parquet write tuning. parquetio.WriteOptions exposes RowGroupRows (predicate-pushdown-friendly small groups vs. compression-friendly large ones), BloomFilterColumns + BloomFilterFPP (equality-filter skipping in DuckDB / Spark / Polars / pyarrow readers today; gobi's own reader when the query optimizer lands).
  • Parallelism controls. Package-level SetMaxParallelism(n) or per-op Workers(n).
  • Pure Go, no cgo. No GDAL, no GEOS, no libproj. Cross-compiles cleanly to every architecture Go supports.

Install

go get github.com/zoobst/gobi

Requires Go 1.26 or newer.

Docs

Full API reference is on pkg.go.dev — auto-generated from source doc comments. Every subpackage (parquetio, csvio, geometry, geojsonio, gpkgio, pgio, kmlio, shpio) has its own page; use the same base URL with the package path appended.

Quick start

Read a CSV, write a GeoParquet
package main

import (
    "github.com/zoobst/gobi/csvio"
    "github.com/zoobst/gobi/parquetio"
)

type city struct {
    Name       string `csv:"name"`
    Population int64  `csv:"population"`
    Geom       string `csv:"geometry" geom:"true"`
}

func main() {
    // .gz / .zst / .bz2 are auto-detected from the filename; explicit
    // ReadOptions.Compression overrides.
    df, err := csvio.ReadFile[city]("cities.csv.gz", &csvio.ReadOptions{CRSHint: 4326})
    if err != nil { panic(err) }
    defer df.Release()

    // The output file carries a spec-compliant GeoParquet 1.1 metadata
    // blob and reads cleanly in GeoPandas / QGIS. nil = Snappy + arrow's
    // default row-group sizing.
    _ = parquetio.WriteFile(df, "cities.parquet", nil)
}
Build a geometry column from lat/lng
// Turn two numeric columns into a WKB geometry column. Argument
// order is x, y — i.e. longitude, latitude — matching WKB / GeoJSON /
// geopandas.points_from_xy. Nulls on either side yield a null point.
lng, _   := df.Column("lng")
lat, _   := df.Column("lat")
points, _ := gobi.PointsFromXY(lng, lat, 4326)
df, _     = df.WithColumn("geometry", points)
// df["geometry"] is now a proper WKB column, ready for SJoin,
// GeoParquet write, etc.
Spatial join
cities, _ := parquetio.ReadFile("cities.parquet", nil)   // 1M points
regions, _ := parquetio.ReadFile("regions.parquet", nil) // 5k polygons

// Which region contains each city?
joined, err := cities.SJoin(regions, "geometry", "geometry", gobi.SPWithin)

// Cap parallelism per-op (see "Parallelism" below):
joined, err = cities.SJoin(regions, "geometry", "geometry", gobi.SPWithin, gobi.Workers(4))
GroupBy + aggregate
gb, _ := df.GroupBy("region")
totals, _ := gb.Agg(
    gobi.Aggregation{Column: "population", Kind: gobi.AggSum},
    gobi.Aggregation{Column: "population", Kind: gobi.AggMean, Alias: "avg_pop"},
)
Sort
// Multi-key, stable, nulls-last. Earlier keys have priority;
// later keys break ties.
sorted, _ := df.SortBy(
    gobi.SortKey{Column: "date"},                        // ascending
    gobi.SortKey{Column: "revenue", Descending: true},
)
Tuned parquet write
// Small row groups + bloom filters on high-cardinality equality columns.
// DuckDB / Spark / Polars / pyarrow readers all use the bloom filters
// for predicate pushdown on equality filters.
err := parquetio.WriteFile(df, "events.parquet", &parquetio.WriteOptions{
    Codec:              parquetio.CodecZstd,
    RowGroupRows:       128_000,                     // 0 = arrow default (~1M)
    BloomFilterColumns: []string{"user_id", "session_id"},
    BloomFilterFPP:     0.01,                        // 0 = arrow default (0.05)
})
Streaming ETL
// Reads a 5 GB parquet file at ~15 MB peak memory. Only two columns are
// fetched off disk; the rest are never decompressed.
err := parquetio.ReadFileChunksFunc(
    "events.parquet",
    &parquetio.ReadOptions{Columns: []string{"user_id", "ts"}, ChunkRows: 64_000},
    func(batch *gobi.Frame) error {
        // Process ~64k rows at a time. The batch is released after
        // return; call batch.Retain() to keep it past this callback.
        return sink.Write(batch)
    },
)

CSV has the same shape: csvio.ReadFileChunksFunc[Row](path, opts, fn).

Derived columns

Two shapes. WithColumn accepts any Series the caller built by hand:

// A user-space helper produces the derived Series any way it likes
// (external library call, vectorized loop, whatever). WithColumn wires
// it back into the Frame — appending, or replacing an existing column
// of the same name.
lat, _   := df.Column("lat")
lng, _   := df.Column("lng")
cells, _ := h3x.Encode(lat, lng, 9)
df, _     = df.WithColumn("h3", cells)

df, _ = df.DropColumn("raw_geometry")

WithColumnExpr accepts an expression tree, so pipelines composed from built-in ops read left-to-right:

// Same shape via the expression IR — no intermediate Series to name.
df, _ = df.WithColumnExpr("usd_price",
    gobi.Col("eur_price").Mul(gobi.Lit(1.08)),
)
df, _ = df.WithColumnExpr("margin",
    gobi.Col("revenue").Sub(gobi.Col("cost")),
)
User-defined aggregation
// Compute the 95th percentile of a numeric column per group.
type P95 struct{}

func (P95) Aggregate(s gobi.Series, rows []int) (any, error) {
    arr := s.Column().Data().Chunks()[0].(*array.Float64)
    vals := make([]float64, 0, len(rows))
    for _, r := range rows {
        if !arr.IsNull(r) { vals = append(vals, arr.Value(r)) }
    }
    if len(vals) == 0 { return nil, nil }
    sort.Float64s(vals)
    return vals[int(float64(len(vals)-1)*0.95)], nil
}
func (P95) Type() arrow.DataType { return arrow.PrimitiveTypes.Float64 }
func (P95) Name() string          { return "p95" }

// Mix custom + built-in aggregations in one call.
gb, _ := df.GroupBy("h3")   // Uint64 keys are hashable
out, _ := gb.Agg(
    gobi.Aggregation{Column: "latency_ms", Kind: gobi.AggMean},
    gobi.Aggregation{Column: "latency_ms", Fn: P95{}},
)
Filter

Two shapes. Filter takes an already-computed Boolean Series (mask):

pops, _ := df.Column("population")
mask, _ := pops.GtScalar(1_000_000)
big,  _ := df.Filter(mask)

FilterExpr takes an expression tree — no intermediate Series to name:

big, _ := df.FilterExpr(
    gobi.Col("population").Gt(gobi.Lit(1_000_000)).
        And(gobi.Col("country").Eq(gobi.Lit("US"))),
)
Reproject
p := geometry.Point{X: -73.9857, Y: 40.7484, CRSValue: geometry.WGS84}
utm, _ := p.EstimateUTMCRS()   // EPSG:32618 (WGS 84 / UTM zone 18N)
proj,  _ := p.ToCRS(utm)       // coordinates now in meters
Buffer + simplify
poly := geometry.SimplePolygon(points, geometry.PseudoMercator)
buffered := poly.Buffer(100, 32) // 100-unit buffer, 32-segment circle approx
simpler  := poly.Simplify(5.0)   // Douglas-Peucker at 5-unit tolerance
R-tree
tree := geometry.NewRTree(bboxes)
hits    := tree.Search(query)      // IDs whose bounds intersect query
nearest := tree.Nearest(x, y, k)   // k closest bounds, sorted
Datetime + timezone
type event struct {
    Name string    `csv:"name"`
    When time.Time `csv:"when" time:"2006-01-02 15:04:05"`
}

df, _ := csvio.ReadFile[event]("events.csv", nil)
when, _ := df.Column("when")

// Render the same instants in New York local time.
nyWhen, _ := when.WithTimezone("America/New_York")

// Component extractors honor the tz.
hourNY, _ := nyWhen.Hour()   // Int64 series with local hours

// Truncate to the top of each local day.
dayStart, _ := nyWhen.TruncateToCalendar(gobi.CalendarDay)
Resample + rolling
// Downsample to hourly buckets (Unix-epoch aligned).
r, _ := df.ResampleEvery("when", time.Hour)
hourly, _ := r.Agg(
    gobi.Aggregation{Column: "value", Kind: gobi.AggSum},
    gobi.Aggregation{Column: "value", Kind: gobi.AggMean, Alias: "avg"},
)

// Trailing 5-minute rolling sum keyed by timestamp.
tr, _ := df.RollingBy("when", 5*time.Minute)
rollSum, _ := tr.Agg("value", gobi.AggSum)

// Fixed-window rolling on a plain Series.
val, _ := df.Column("value")
m7, _ := val.RollingMean(7) // 7-row moving average
KML / Shapefile
// KML → Frame (auto-parses ExtendedData into columns)
places, _ := kmlio.ReadFile("places.kml")
_ = kmlio.WriteFile(places, "out.kml")

// Shapefile → Frame (reads .shp + .shx + .dbf + optional .prj)
counties, _ := shpio.ReadFile("counties")           // no .shp suffix needed
_ = shpio.WriteFile(counties, "counties_out")       // writes all four files

Packages

Package What it does
github.com/zoobst/gobi Frame, Series, GroupBy, Join, SJoin, Explode, datetime + rolling + resample, options
.../gobi/geometry 2D + XYZ primitives, WKB / WKT, CRS + reprojection, predicates, R-tree, Buffer / Simplify / Centroid
.../gobi/csvio Typed CSV read + streaming (ReadFileChunksFunc), gzip / zstd / bzip2 auto-detect
.../gobi/parquetio Parquet read/write + streaming + column projection + row-group + bloom-filter tuning; snappy/gzip/brotli/zstd/lz4 + GeoParquet 1.1
.../gobi/geojsonio Full RFC 7946 GeoJSON (all geometry types + XYZ) — Frame-level ReadFile/WriteFile/ScanFile, .geojsonl streaming
.../gobi/gpkgio Read / write OGC GeoPackage 1.3 (SQLite) with RTree spatial index + LazyFrame ScanFile + SQL predicate pushdown
.../gobi/pgio Beta. PostgreSQL / PostGIS via pgx/v5 — ReadQuery/ReadTable/ScanTable + WriteTable with CopyFrom bulk load. Integration tests are //go:build integration-gated; set PGIO_TEST_DSN and run against a live PostGIS to exercise them.
.../gobi/kmlio Read / write KML (OGC 12-007r2) with Placemarks + ExtendedData
.../gobi/shpio Read / write ESRI Shapefile (.shp + .shx + .dbf + optional .prj)

Geometry columns

gobi intentionally does not use an Arrow custom-extension type for geometry. Geometries are Arrow Binary columns holding WKB, with the column marked in schema metadata:

"gobi:geometry_type" = "WKB"
"gobi:crs_epsg"      = "4326"

When writing Parquet, gobi additionally emits a proper GeoParquet 1.1 geo blob at the file level with primary_column, geometry_types, crs, and bbox. That is what makes gobi-produced files interoperate with GeoPandas and QGIS out of the box.

Use gobi.GeometryField(name, epsg) to construct a tagged field manually.

Extending gobi

Three extension points cover most add-on work today, without forking:

1. Derived columns via a helper package + Frame.WithColumn. Write a sibling package (e.g. h3x, hashcol) whose functions take one or more gobi.Series and return a gobi.Series. Users compose:

lat, _   := df.Column("lat")
lng, _   := df.Column("lng")
cells, _ := h3x.Encode(lat, lng, 9)
df, _     = df.WithColumn("h3", cells)

Because the helper controls the whole loop, it can dispatch to native libraries (H3, MurmurHash, whatever) once per row without the DataFrame having to know anything about the operation. WithColumn appends or replaces, DropColumn removes.

2. Custom aggregations via the Aggregator interface.

type Aggregator interface {
    // Reduce s[rows...] to a single scalar. Return nil for a null.
    Aggregate(s Series, rows []int) (any, error)
    // Declares the arrow type of Aggregate's return values. Supports
    // Float32/64, Int32/64, Uint32/64, Bool, String, Binary, Timestamp.
    Type() arrow.DataType
    // Suffix for the default output column name.
    Name() string
}

Set Aggregation{Column: "col", Fn: myAgg} and call GroupBy.Agg as usual. Mix custom + built-in aggregations in a single call. If the returned dynamic type doesn't match the declared Type(), Agg returns an error naming the offending aggregation rather than panicking.

3. Custom expression nodes via the ExprNode interface.

type ExprNode interface {
    // Evaluate against a Frame. Return a Series with input.NumRows() rows.
    Eval(input *Frame) (Series, error)
    // Declared output arrow type given the input schema. Used by
    // FilterExpr / WithColumnExpr for validation.
    Type(schema *arrow.Schema) (arrow.DataType, error)
    // Sub-expressions, for tree walkers.
    Children() []Expr
    // Pretty-printer for logs and debug output.
    String() string
}

Wrap your node with gobi.Custom(node) and it composes with the built-in Col, Lit, and operator methods:

// h3x.Encode returns a gobi.Expr backed by a custom node.
cellExpr := h3x.Encode(gobi.Col("lat"), gobi.Col("lng"), 9)

df, _ = df.WithColumnExpr("h3", cellExpr)
df, _ = df.FilterExpr(cellExpr.Eq(gobi.Lit(uint64(0xdead))))

Because expressions are data, not function calls, the tree can be inspected before evaluation (e.String(), e.Node().Children()), type-checked without touching the buffers (e.Node().Type(schema)), and — in a future release — rewritten by an optimizer that pushes predicates into scans and prunes unused columns. Extension points that implement ExprNode will benefit from those passes automatically.

Group-by key types. Hashable key columns: String, Bool, Int32, Int64, Uint32, Uint64, Float64, Timestamp. Uint64 is what makes H3-cell grouping ergonomic.

Parallelism

Two layers of parallel execution work together in a LazyFrame collect: the parquet scan splits row-groups across workers, and the streaming aggregate partitions rows by key hash across workers.

  • Parallel scan. parquetio.ScanFile(path, &parquetio.ReadOptions{ScanWorkers: N}) splits row-groups across N goroutines. Each worker reads a disjoint subset of row-groups; batches fan-in through a bounded channel. ScanWorkers: 0 (the default) auto-picks GOMAXPROCS, capped at the file's row-group count. ScanWorkers: 1 forces serial for reproducibility. Files with a single row-group skip parallel scan automatically (no benefit possible).
  • Parallel aggregate. The streaming hash aggregate (GroupBy(...).Agg(...) on a LazyFrame with built-in Kinds) partitions rows across GOMAXPROCS workers by key hash — no cross-worker key overlap, no locks, no value-level combine at merge. Kicks in for any aggregate where every Aggregation uses a built-in Kind; custom Fn aggregators still route through the materializing fallback.

Both layers respect the package-level SetMaxParallelism(n) / per-op Workers(n) overrides, in this priority:

  1. Per-op gobi.Workers(n) option
  2. Package default via gobi.SetMaxParallelism(n)
  3. GOMAXPROCS
gobi.SetMaxParallelism(4)                 // process-wide default
df.SJoin(..., gobi.Workers(8))            // override for one call
df.SJoin(..., gobi.Workers(1))            // force sequential

LazyFrame.ExplainPhysical() shows the resolved worker count for each parallel node — useful when debugging why a query didn't get the parallelism you expected.

Design constraints

  • Pure Go, no cgo. GDAL, GEOS, libproj, and other C libraries are intentionally off the table. This keeps go build clean across every platform Go targets and avoids the LGPL/toolchain overhead. The trade: no File Geodatabase support, no polygon Union/Intersection/Difference (would require a Vatti / Martinez-Rueda hand-roll), no PROJ-grade reprojection beyond WGS84 / Web Mercator / UTM.

Performance

All numbers Apple M3 Pro, warm cache, 10–20 iterations per op. Fixtures and scripts live under benchmarks/ — regenerate with go run generate_fixture.go, go run generate_csv_fixture.go, and go run generate_spatial_fixture.go.

Compute ops (1M-row Parquet, non-spatial)
Op gobi pandas 2.3 Polars 1T Polars all
Sum(value_a) 0.92 ms 0.31 ms 0.08 ms 0.08 ms
value_a + value_b 1.01 ms 0.71 ms 0.68 ms 0.76 ms
Filter(value_a > 500k) 14.3 ms 4.81 ms 1.74 ms 1.37 ms
GroupBy(key).Agg(Sum,Mean) 33.3 ms 18.16 ms 7.59 ms 2.24 ms

Polars 1T = POLARS_MAX_THREADS=1; Polars all = default (all cores). pandas sits between gobi and single-threaded Polars on every op — numpy's SIMD reductions and C-implemented groupby carry it past pure-Go for now. See the SIMD note below.

CSV read (38.6 MB / 1M rows)
Reader per-read notes
Polars 1.42, all threads 9.1 ms multi-threaded Rust tokenizer + SIMD (typed schema)
Polars 1.42, 1 thread 56.0 ms SIMD numeric parse, single core
pandas 2.3, engine="pyarrow" 25.9 ms pyarrow C++ tokenizer
pandas 2.3, default (C engine) 129.8 ms pandas' native C tokenizer
gobi csvio.Read 209.4 ms arrow-go's CSV wraps stdlib encoding/csv

The gap is entirely in stdlib encoding/csv allocating a []string per row + per-cell strconv — 99.5% of gobi's CSV allocations show up there in a pprof run. Closing it means replacing that layer with a byte-level tokenizer that writes straight into Arrow buffers; not on the roadmap yet. Maybe the arrow-go folks will pick that up.

Spatial ops (100k points × 100 polygons)
Op gobi geopandas 1.1 result
Read points.parquet (100k) 4.37 ms 35.2 ms 8.0× faster
Read polygons.parquet (100) 0.26 ms 0.76 ms 2.9× faster
Area(polygons) 0.02 ms 0.13 ms 6.5× faster
Centroid(polygons) 0.02 ms 0.16 ms 8× faster
SJoin(100k pts, 100 polys) 3.14 ms 2.60 ms 1.2× slower

Gobi wins on read and per-row bulk ops because it doesn't have to construct Shapely Python objects per row on load. The one gap is sjoin: geopandas uses Shapely 2's GEOS-backed STRtree in C++; gobi's Sort-Tile-Recursive R-tree is pure Go. Landing within 40% of a GEOS-C++ index while staying cgo-free is the intended trade.

LazyFrame + optimizer (projection pushdown)

Same 1M-row parquet fixture, Select(id, value_a) — reading 2 of 4 columns. Measures the projection-pushdown rule's effect on I/O and decode cost.

Path per-op vs. baseline
ReadFile(path, Options{Columns:[id,value_a]}) (eager) 6.18 ms 1.0× (baseline)
ScanFile(path).Select(id,value_a).Collect() (optimized) 7.06 ms ~1.14× — close to eager
ScanFile(path).Select(id,value_a).CollectRaw() (no rules) 14.08 ms 2.3× slower — reads all 4 cols

The optimizer's ProjectionPushdown rule turns the lazy pipeline into the equivalent of an eager Options.Columns — 2.0× faster than the same pipeline with optimization disabled. Optimizer overhead itself is 8 µs per plan (measured on a 5-node pipeline), or 0.14% of the collect time. Always-on optimization is effectively free.

1 Billion Row Challenge

The 1BRC fixture is 1 billion weather-station rows in Snappy-compressed parquet (~4 GB on disk). Query: min / mean / max of temperature grouped by station. Apple M3 Pro, 11 GOMAXPROCS.

Config wall user CPU peak RSS
Serial (ScanWorkers=1, single agg) 1m 50s 147s 156 MB
Parallel scan 1m 52s 148s 1.28 GB
Parallel scan + parallel aggregate. 1m 11s 197s 1.31 GB
Both + composite-key optimizations 15.5s 141s 1.5 GB
Polars 1.42 streaming (reference) ~3 s ~15s ~4.6 GB

Same query, streaming end-to-end — no LazyFrame.CollectRaw() materialization, no disk spill. The gap to Polars closed to 5× across three complementary changes: partitioning the parquet scan across row-groups, sharding the streaming hash aggregate by key hash across workers, and eliminating per-row key allocations in the hot path (reusable scratch buffers + single-string-key fast path that reads the arrow value zero-copy).

Peak RSS is 3× lower than Polars because gobi keeps at most one batch per worker in memory (~1.3 GB total on 11 workers) — Polars buffers larger working sets by design.

Why the remaining compute-op gap will shrink

Sum / Add are already memory-bandwidth-bound. The remaining gap on Sum is SIMD reduction (Polars and numpy both use parallel-lane accumulators). The Go simd and simd/archsimd packages gain arm64 NEON support in Go 1.27 (August 2026), at which point the existing //go:build goexperiment.simd kernel gets rewritten against the portable package and closes most of the Sum gap — see TODO(1.27-simd) in series_ops_simd_*.go.

The 1BRC gap breaks down similarly: with parallelism landed, most of the remaining 5× vs Polars is per-row accumulator throughput (minMaxAcc.Update calls Series.numericAt per row, dispatching through an interface + type switch). Vectorized kernels — tight loops over typed slices — would close a meaningful fraction on the Go compiler alone; the last factor comes from the same Go 1.27 SIMD package. Neither is on the roadmap until the toolchain support lands.

Development

  • go test -race ./... should pass before you push.
  • Please keep dependencies minimal — Arrow is the one big one on purpose.

License

MIT. See LICENSE.

Documentation

Overview

Package gobi is a geospatial dataframe library for Go, built on Apache Arrow.

A DataFrame is a set of named, equal-length Series backed by Arrow columns. Geometry columns are stored as Well-Known Binary (WKB) in Arrow Binary arrays and marked as geometries in the schema metadata (following the GeoParquet convention). This keeps gobi interoperable with other Arrow- aware tools without a custom extension type registry.

Reading data 

df, err := gobi.ReadCSV("cities.csv", nil)
df, err := gobi.ReadParquet("cities.parquet")

Selecting columns and rows 

col, _ := df.Column("name")
head, _ := df.Head(10)

Working with geometries 

geom, _ := df.Geometry("geometry", 0) // decode the WKB at row 0

See the subpackages for more:

  • geometry: 2D primitives, WKB/WKT, CRS, area/distance/hull
  • csvio: typed CSV read/write
  • parquetio: Parquet + GeoParquet read/write, LazyFrame scan
  • geojsonio: GeoJSON encoding/decoding for individual features
  • gpkgio: OGC GeoPackage read/write with RTree spatial index
  • kmlio: KML read/write with ExtendedData attributes
  • shpio: ESRI Shapefile read/write

Index

Constants

View Source 
const (
	GeoParquetVersion = "1.1.0"
	// GeoParquetMetadataKey is the file-level metadata key used by GeoParquet.
	GeoParquetMetadataKey = "geo"
)

GeoParquetVersion is the GeoParquet spec version this package emits.

View Source 
const (
	MetaGeometryType = "gobi:geometry_type"
	MetaGeometryCRS  = "gobi:crs_epsg"
)

Schema metadata keys used by gobi to tag geometry columns. This follows the GeoParquet convention closely enough that files written by gobi should be readable by other GeoParquet-aware tools for primitive geometry types.

View Source 
const SJoinMinParallelRows = 1024

SJoinMinParallelRows is the smallest left-frame size at which SJoin will spawn goroutines. Below this, the sequential path is used to avoid scheduling overhead swamping the useful work.

Variables

View Source 
var (
	ErrColumnNotFound     = errors.New("gobi: column not found")
	ErrColumnLenMismatch  = errors.New("gobi: column length mismatch")
	ErrColumnTypeMismatch = errors.New("gobi: column type mismatch")
	ErrRowOutOfRange      = errors.New("gobi: row index out of range")
	ErrEmptyFrame         = errors.New("gobi: empty dataframe")
	ErrNotGeometry        = errors.New("gobi: column is not a geometry column")
	ErrExprTypeMismatch   = errors.New("gobi: expression type mismatch")
	ErrUnsupportedLiteral = errors.New("gobi: unsupported literal type")
)
View Source 
var ErrMaskNotBoolean = fmt.Errorf("gobi: filter mask must be a boolean series")

ErrMaskNotBoolean is returned when Filter receives a non-boolean Series.

View Source 
var ErrNotDateTime = fmt.Errorf("gobi: series is not a datetime column")

ErrNotDateTime is returned when a time-only op is attempted on a Series that is not a Timestamp / Date32 / Date64 column.

View Source 
var ErrNotMonotonic = fmt.Errorf("gobi: time column must be monotonically non-decreasing")

ErrNotMonotonic is returned by ops that require a time column sorted in non-decreasing order (currently: Frame.RollingBy).

View Source 
var ErrNotNumeric = fmt.Errorf("gobi: series is not numeric")

ErrNotNumeric is returned when arithmetic is attempted on a non-numeric Series. Only int64/int32/float64/float32 are currently supported.

Functions

func CanPossiblyMatch 

func CanPossiblyMatch(pred Expr, stats Stats) bool

CanPossiblyMatch reports whether pred could be satisfied by any row in the range described by stats. Returns true when uncertain — false positives are safe (over-read); false negatives break correctness.

Supported predicate shapes:

  • AND / OR of shapes below
  • col == literal, col != literal
  • col <, <=, >, >= literal
  • literal on either side (auto-normalized)

Anything else (NOT, arithmetic, custom nodes) is treated as "possibly matches." Used by parquetio for row-group skipping; callable by any source package that can produce column stats.

func ExprToSQL 

func ExprToSQL(e Expr) (string, []any, bool)

ExprToSQL translates a gobi.Expr into a SQL fragment + a slice of bind arguments suitable for parameterized SQL queries (SQLite, PostgreSQL / PostGIS, DuckDB, etc.).

Returns ok=false when the expression contains a node the translator can't represent: custom nodes (Custom), aliased sub-expressions inside a predicate, or any binary op that doesn't map cleanly to SQL. Callers that get ok=false should fall back to evaluating the expression in the executor rather than pushing it into SQL.

The emitted SQL uses positional `?` placeholders for every literal, matching the SQLite driver contract that gpkgio uses. Drivers that expect `$1`/`$2` (pgx) or `:name` (Oracle) can rewrite the placeholders trivially since the args slice is 1:1 with `?` order.

Column names are double-quoted per SQL:2016 identifier rules; any embedded double-quote in a name is escaped by doubling. NULL comparisons (`x = NULL`) are rewritten to `IS NULL` / `IS NOT NULL` to match SQL's null-safe semantics.

Example: 

e := gobi.Col("price").Mul(gobi.Lit(1.08)).Gt(gobi.Lit(100.0))
sql, args, ok := gobi.ExprToSQL(e)
// sql  = `(("price" * ?) > ?)`
// args = []any{1.08, 100.0}
// ok   = true

func GeoParquetSchemaWithMetadata 

func GeoParquetSchemaWithMetadata(schema *arrow.Schema, meta *GeoParquetMetadata) (*arrow.Schema, error)

GeoParquetSchemaWithMetadata returns a copy of schema with the given GeoParquet metadata injected under the "geo" key at the file level.

func GeometryField 

func GeometryField(name string, epsg int32) arrow.Field

GeometryField returns a schema field tagged as a WKB geometry column. Pass epsg=0 to leave the CRS unset.

func MarshalGeoParquetMetadata 

func MarshalGeoParquetMetadata(meta *GeoParquetMetadata) (string, error)

MarshalGeoParquetMetadata serializes meta to JSON.

func MaxParallelism 

func MaxParallelism() int

MaxParallelism returns the current package-level default worker count. Returns 0 if unset (in which case operations use GOMAXPROCS).

func SetMaxParallelism 

func SetMaxParallelism(n int)

SetMaxParallelism sets the default max worker count for every parallel gobi operation across the process. Pass 0 (the default) to defer to GOMAXPROCS. Individual calls can override with the Workers option.

Negative values are treated as zero.

Types

type AggKind 

type AggKind uint8

AggKind identifies an aggregation operation. 

const (
	AggCount AggKind = iota
	AggSum
	AggMean
	AggMin
	AggMax
	// AggFirst / AggLast: first (or last) non-null value in the
	// group, in the group's row order. Output type matches the source
	// column (preserved via the Frame's schema).
	AggFirst
	AggLast
	// AggStd / AggVar: sample standard deviation / variance
	// (Bessel-corrected, n-1 denominator). Streaming path uses
	// Welford's online algorithm. Empty or single-row groups emit
	// null.
	AggStd
	AggVar
	// AggNUnique: count of distinct non-null values per group. Output
	// is Int64, always non-null (empty group counts as 0).
	AggNUnique
)

func (AggKind) String 

func (k AggKind) String() string

type Aggregation 

type Aggregation struct {
	Column string
	Kind   AggKind
	Alias  string
	Fn     Aggregator
}

Aggregation names a column to aggregate and how to aggregate it. The resulting column in the aggregated frame is named Alias (or, if empty, "<column>_<kind>" for built-in kinds, "<column>_<Fn.Name()>" for custom aggregators).

When Fn is non-nil, it takes precedence over Kind: the aggregation is user-defined and Fn is called once per group. Otherwise Kind selects a built-in aggregation.

type Aggregator 

type Aggregator interface {
	Aggregate(s Series, rows []int) (any, error)
	Type() arrow.DataType
	Name() string
}

Aggregator is a user-defined aggregation function called once per group during GroupBy.Agg. Implementations reduce the rows of a Series to a single scalar value.

Typical implementations: 

weighted mean, mode, percentile / quantile, log-sum-exp, first / last,
geospatial reductions (H3 cell of the centroid, dominant hex),
string aggregations (concat, longest common prefix).

Return nil from Aggregate to emit an Arrow null for the group. The declared Type must match the concrete Go type Aggregate returns: 

Float64 → float64        Uint64  → uint64      String    → string
Float32 → float32        Uint32  → uint32      Binary    → []byte
Int64   → int64          Boolean → bool        Timestamp → arrow.Timestamp
Int32   → int32

If the returned dynamic type doesn't match Type, Agg reports an error naming the offending Aggregation.

Aggregate is called sequentially per group; the same Aggregator instance is reused across groups. Implementations that need per-group scratch space should allocate it inside Aggregate rather than as receiver fields.

type CalendarUnit 

type CalendarUnit uint8

CalendarUnit selects a calendar-aware truncation granularity for Series.TruncateToCalendar. 

const (
	CalendarWeek  CalendarUnit = iota // ISO week: truncate to the previous Monday 00:00
	CalendarMonth                     // truncate to the first day of the month at 00:00
	CalendarYear                      // truncate to Jan 1 at 00:00
)

type ExecOperator 

type ExecOperator interface {
	// Next returns the next batch. When no more batches are available
	// it returns (nil, io.EOF). Callers must Release the returned
	// batch when done.
	Next(ctx context.Context) (arrow.RecordBatch, error)

	// Schema is the operator's output schema. Stable across all
	// batches; safe to call before Next has been called.
	Schema() *arrow.Schema

	// Close releases resources held by the operator AND its upstream
	// operators. Idempotent — safe to call multiple times. Called by
	// Execute at the end of a plan, so hand-written test drivers
	// usually don't need to call it explicitly.
	Close() error
}

ExecOperator is a pull-based iterator over arrow record batches. Each call to Next produces the next batch of rows (or io.EOF when the operator has exhausted its input); the returned batch is owned by the caller and must be Released when no longer needed.

Operators can be composed into a tree: `filterExec` wraps another ExecOperator as its input and yields filtered batches; `scanFrameExec` is a leaf that produces batches from a source Frame. The plan compiler (Compile) turns a LogicalPlan into such a tree.

Layer 6 (vectorized execution) is the migration from the whole- Frame-per-node dispatch in lazy.go's collectPlan to batch-streaming through operators. Slice 1 lands the interface + streaming filter / project / limit / with-column / drop and a materializing fallback for blocking ops (Sort, Aggregate, Join, Tail); those still route to the eager engine internally but see one input Frame instead of a per-step re-materialization.

func Compile 

func Compile(p LogicalPlan) (ExecOperator, error)

Compile translates a LogicalPlan tree into a tree of ExecOperators ready for streaming execution via Execute.

The mapping is one-to-one for streaming operators (Filter → filterExec, Project → projectExec, and so on). Blocking operators (Sort, Aggregate, Join, Tail) compile to a materializeExec that pulls its upstream to a Frame and delegates the actual computation to the existing eager engine. This keeps Layer 6 slice 1 focused on the execution model itself; later slices can replace each materializeExec with a native streaming implementation.

Compile itself does no I/O and starts no goroutines. Scan operators that use a background producer (scanFileExec) start their goroutine on construction, so callers should always follow Compile with Execute or the operator's Close to avoid leaks.

type Expr 

type Expr struct {
	// contains filtered or unexported fields
}

Expr is an expression tree — a value that describes a computation without performing it. Build expressions with the package-level constructors (Col, Lit, Custom) and combine them with the fluent methods on Expr (Add, Mul, Gt, And, etc.).

Expressions evaluate lazily: nothing runs until Frame.FilterExpr or Frame.WithColumnExpr consumes them. Because the tree is data, it can be inspected, printed, and (in a future release) rewritten by an optimizer before evaluation.

Example: 

// (price * 1.08) > 100
e := gobi.Col("price").Mul(gobi.Lit(1.08)).Gt(gobi.Lit(100.0))
filtered, err := df.FilterExpr(e)

func Col 

func Col(name string) Expr

Col returns an expression that references the column named name in the input Frame.

func Custom 

func Custom(node ExprNode) Expr

Custom wraps a user-defined ExprNode into an Expr. This is the entry point for extending gobi with your own expression types.

func Lit 

func Lit(v any) Expr

Lit returns an expression that evaluates to a constant value. The value's arrow type is inferred from its Go type; the following are supported: 

bool                → Boolean
int, int32, int64   → Int64
float32, float64    → Float64
string              → String

Other Go types return an Expr whose Eval reports a type-inference error.

func SplitConjuncts 

func SplitConjuncts(e Expr) []Expr

SplitConjuncts breaks an expression at top-level AND boundaries so callers can push the pieces that translate to SQL individually and keep the rest for post-scan filtering.

Example: given `(a > 5 AND custom_fn(b) AND c = "x")`, SplitConjuncts yields `[a > 5, custom_fn(b), c = "x"]`. The custom_fn part won't translate; the other two will, and the caller can push `a > 5 AND c = "x"` into SQL while leaving `custom_fn(b)` in the executor.

Non-AND expressions come back as a single-element slice — a top-level OR isn't safe to split (both sides must match), so we don't try.

func (Expr) Add 

func (e Expr) Add(o Expr) Expr

Add returns e + o.

func (Expr) Alias 

func (e Expr) Alias(name string) Expr

Alias renames the output column produced by e. Only used by Frame.WithColumnExpr and future SelectExpr — has no effect inside FilterExpr.

func (Expr) And 

func (e Expr) And(o Expr) Expr

And returns e AND o. Both e and o must produce Boolean values.

func (Expr) Div 

func (e Expr) Div(o Expr) Expr

Div returns e / o.

func (Expr) Eq 

func (e Expr) Eq(o Expr) Expr

Eq returns e == o (Boolean result).

func (Expr) Ge 

func (e Expr) Ge(o Expr) Expr

Ge returns e >= o (Boolean result).

func (Expr) Gt 

func (e Expr) Gt(o Expr) Expr

Gt returns e > o (Boolean result).

func (Expr) Le 

func (e Expr) Le(o Expr) Expr

Le returns e <= o (Boolean result).

func (Expr) Lt 

func (e Expr) Lt(o Expr) Expr

Lt returns e < o (Boolean result).

func (Expr) Mul 

func (e Expr) Mul(o Expr) Expr

Mul returns e * o.

func (Expr) Ne 

func (e Expr) Ne(o Expr) Expr

Ne returns e != o (Boolean result).

func (Expr) Node 

func (e Expr) Node() ExprNode

Node returns the underlying ExprNode. Rarely needed unless you're implementing your own tree walker or optimizer.

func (Expr) Not 

func (e Expr) Not() Expr

Not returns the logical negation of e. e must produce Boolean values.

func (Expr) Or 

func (e Expr) Or(o Expr) Expr

Or returns e OR o. Both e and o must produce Boolean values.

func (Expr) String 

func (e Expr) String() string

String returns a human-readable representation of the expression.

func (Expr) Sub 

func (e Expr) Sub(o Expr) Expr

Sub returns e - o.

type ExprNode 

type ExprNode interface {
	// Eval evaluates the expression against input, returning a Series
	// whose values are the result. The returned Series should have
	// length equal to input.NumRows(). Scalar broadcasts are handled
	// by the caller for built-in nodes; custom nodes may either
	// broadcast internally or emit a length-1 Series.
	Eval(input *Frame) (Series, error)

	// Type returns the arrow data type of the result given the input
	// schema. Called by type-inference passes before evaluation.
	Type(schema *arrow.Schema) (arrow.DataType, error)

	// Children returns the sub-expressions of this node in
	// deterministic order (empty for leaves). Used by tree walkers.
	Children() []Expr

	// String returns a human-readable representation of the node.
	String() string
}

ExprNode is the interface every expression node implements. Users extending gobi with custom expression types (H3 encoding, hashes, ML inference, etc.) implement this and wrap the result with Custom.

type Frame 

type Frame struct {
	// contains filtered or unexported fields
}

Frame is a columnar dataset: an Arrow schema plus a set of named Series.

The API mirrors GeoPandas / Polars in shape: Shape, Head, Tail, Row, Column, and geometry helpers.

func Concat 

func Concat(frames ...*Frame) (*Frame, error)

Concat is the package-level counterpart to Frame.Concat: stacks rows from every frame in-order, no dedupe. Useful when the caller holds a []*Frame and doesn't want the `frames[0].Concat(frames[1:]...)` dance the method form requires.

Errors when frames is empty (there's nothing to return a schema from) or when any two frames have incompatible schemas — same rules as Frame.Concat.

func Execute 

func Execute(ctx context.Context, op ExecOperator) (*Frame, error)

Execute drives op to EOF, gathers every batch, and assembles them into a single-chunk *Frame. Closes op unconditionally.

This is the terminal entry point for the streaming executor: LazyFrame.Collect calls Compile + Execute. Callers that want to consume batches directly (streaming ETL) should walk op.Next in a loop themselves — but that's a lower-level API not yet exposed.

func NewFrame 

func NewFrame(schema *arrow.Schema, cols []arrow.Column) (*Frame, error)

NewFrame builds a Frame from a schema and a set of arrow columns. The number of columns must match the schema.

func NewFrameFromTable 

func NewFrameFromTable(t arrow.Table) *Frame

NewFrameFromTable adopts the columns of t.

func (*Frame) Column 

func (f *Frame) Column(name string) (Series, error)

Column returns the Series named name.

func (*Frame) ColumnAt 

func (f *Frame) ColumnAt(i int) (Series, error)

ColumnAt returns the Series at position i.

func (*Frame) ColumnNames 

func (f *Frame) ColumnNames() []string

ColumnNames returns the column names in order.

func (*Frame) Concat 

func (f *Frame) Concat(others ...*Frame) (*Frame, error)

Concat returns a new Frame with rows from f followed by the rows of each frame in others, in the order given. No deduplication — this is a pure row stack (SQL UNION ALL / polars vstack / pandas `concat(axis=0)`).

All frames must share the exact same schema: matching column count, order, names, and arrow types. Fingerprint mismatches return an error naming the offending column with both types so the caller can decide how to cast.

Each frame's Arrow buffers are kept alive as separate chunks of the output — no memcpy of the underlying data. The resulting Frame's columns are multi-chunk; downstream ops that assume single-chunk (like some numeric fast paths) will fall back to the general path. Call `.Coalesce()` (planned) if you need a single-chunk output.

func (*Frame) Difference 

func (f *Frame) Difference(other *Frame, cols ...string) (*Frame, error)

Difference returns rows in f that do not appear in other, deduplicated over cols. cols nil/empty → all columns. Nulls are treated as equal.

Equivalent to SQL EXCEPT / polars' `df.filter(~col.is_in(other))` generalized to whole-row membership.

Both frames must share the exact same schema.

func (*Frame) DropColumn 

func (f *Frame) DropColumn(name string) (*Frame, error)

DropColumn returns a new Frame with the named column removed. Returns ErrColumnNotFound if no column matches. Every retained column is independently ref-counted so the caller can Release either Frame without affecting the other's buffers.

func (*Frame) Explode 

func (f *Frame) Explode(geomCol string) (*Frame, error)

Explode returns a new Frame where each multi-geometry row of geomCol has been split into one row per component. Single geometries (Point, LineString, Polygon) pass through unchanged. Null geometries are kept (one output row per null, unchanged). GeometryCollection is expanded to one row per contained geometry.

All non-geometry columns are duplicated across the exploded rows so per-row attributes propagate to every component.

func (*Frame) Filter 

func (f *Frame) Filter(mask Series) (*Frame, error)

Filter returns a new Frame containing only the rows where mask is true. Null mask entries are treated as false.

The mask length must equal the frame's row count.

func (*Frame) FilterExpr 

func (f *Frame) FilterExpr(e Expr) (*Frame, error)

FilterExpr returns a new Frame containing the rows for which e evaluates to true. e must produce a Boolean Series; null entries in the mask are treated as false (SQL-style — matches Frame.Filter).

Example: 

// Keep rows where (price * 1.08) > 100.
e := gobi.Col("price").Mul(gobi.Lit(1.08)).Gt(gobi.Lit(100.0))
out, err := df.FilterExpr(e)

func (*Frame) Geometry 

func (f *Frame) Geometry(colName string, row int) (any, error)

Geometry decodes the geometry at (row, colName). Returns ErrNotGeometry if the column is not a geometry column.

func (*Frame) GroupBy 

func (f *Frame) GroupBy(keys ...string) (*GroupBy, error)

GroupBy returns a GroupBy over the given key column names.

func (*Frame) Head 

func (f *Frame) Head(n int) *Frame

Head returns a Frame with the first n rows (default 5).

func (*Frame) Intersect 

func (f *Frame) Intersect(other *Frame, cols ...string) (*Frame, error)

Intersect returns rows that exist in both f and other, deduplicated over cols. cols nil/empty → all columns. Nulls are treated as equal (see Union for the semantics rationale).

The result draws rows + column values from f (not from other) — so if other has a row with the same key columns but different non-key values, you get f's version. This matches SQL INTERSECT and polars behavior.

Both frames must share the exact same schema.

func (*Frame) Join 

func (f *Frame) Join(right *Frame, leftKey, rightKey string, kind JoinType) (*Frame, error)

Join returns a new Frame formed by combining rows of f (the left frame) and right, where the left column named leftKey equals the right column named rightKey. The join key must be a hashable type (String, Int64, Int32, Bool, Uint32, Uint64, Float64, Timestamp).

The result contains all columns from the left frame followed by all columns from the right frame except the join key. Right-side columns whose names collide with left-side columns are renamed with a "_right" suffix. For JoinSemi and JoinAnti, only left-side columns appear in the output.

Null keys never match (SQL semantics). In JoinLeft, JoinFull, and JoinAnti a null-keyed left row still appears in the output; in JoinRight and JoinFull a null-keyed right row still appears; in JoinSemi and JoinInner it is filtered out.

func (*Frame) Lazy 

func (f *Frame) Lazy() *LazyFrame

Lazy wraps f in a LazyFrame anchored at a Scan[frame] leaf. Fluent operations on the returned LazyFrame append nodes to the plan; Collect() replays them against f.

func (*Frame) NumCols 

func (f *Frame) NumCols() int

NumCols returns the number of columns.

func (*Frame) NumRows 

func (f *Frame) NumRows() int

NumRows returns the number of rows in the frame (0 if there are no columns).

func (*Frame) Pivot 

func (f *Frame) Pivot(index, columns, values string, agg AggKind) (*Frame, error)

Pivot reshapes a long-form Frame into a wide-form one.

  • index: column whose distinct values become the rows.
  • columns: column whose distinct values become the new columns.
  • values: column supplying the cell values.
  • agg: how to reduce when multiple input rows map to the same (index, columns) cell. Use AggFirst / AggLast if you're sure there's no collision; use AggSum / AggMean / etc. to reduce. Any built-in AggKind is accepted.

The output schema is: 

<index column> | <col_value_1> | <col_value_2> | ...

Column values are stringified for use as arrow field names — the header row is always a string. The value column's arrow type follows aggOutputType(Aggregation{Kind: agg, ...}); it's Int64 for Count/NUnique and Float64 for the rest, matching what GroupBy.Agg would produce.

Cells with no matching input rows are emitted as null. Output rows are ordered by the sorted index value; output columns are ordered by the sorted distinct columns-column value so the shape is deterministic across runs.

Equivalent to `pandas.DataFrame.pivot_table(index, columns, values, aggfunc)` / polars' `DataFrame.pivot`. Multi-column pivot indices aren't supported yet — pass a single index column.

func (*Frame) Release 

func (f *Frame) Release()

Release decrements the reference count of the underlying Arrow columns. Callers should match every Retain (including the implicit one at construction) with exactly one Release.

func (*Frame) ResampleEvery 

func (f *Frame) ResampleEvery(timeCol string, interval time.Duration) (*Resampler, error)

ResampleEvery returns a Resampler that will group f's rows into interval-wide buckets by the values in timeCol. The interval must be positive.

func (*Frame) Retain 

func (f *Frame) Retain()

Retain increments the reference count of the underlying Arrow columns.

func (*Frame) RollingBy 

func (f *Frame) RollingBy(timeCol string, period time.Duration) (*TimeRolling, error)

RollingBy returns a TimeRolling. The time column must be a Timestamp series in non-decreasing order.

func (*Frame) Row 

func (f *Frame) Row(i int) (*Frame, error)

Row returns a Frame containing the single row at index i.

func (*Frame) SJoin 

func (f *Frame) SJoin(right *Frame, leftGeomCol, rightGeomCol string, pred SpatialPredicate, opts ...Option) (*Frame, error)

SJoin performs a spatial join of f (left) and right by evaluating pred on each pair of geometries from leftGeomCol and rightGeomCol. Only rows where pred holds are emitted. The output frame contains all columns from the left frame, followed by all columns from the right frame except its geometry column (analogous to Frame.Join dropping the right join key). Right-side column names that collide with left-side names get a "_right" suffix.

Under the hood, an R-tree is built over the right frame's geometry bounds so each left row scans only overlapping candidates. Parallelism follows the priority order documented on resolveWorkers: Workers(n) > package SetMaxParallelism > GOMAXPROCS.

func (*Frame) Schema 

func (f *Frame) Schema() *arrow.Schema

Schema returns the underlying Arrow schema.

func (*Frame) Shape 

func (f *Frame) Shape() (rows, cols int)

Shape returns (rows, cols) — matching Pandas convention.

func (*Frame) SortBy 

func (f *Frame) SortBy(keys ...SortKey) (*Frame, error)

SortBy returns a new Frame with rows arranged according to keys. The sort is stable: rows that compare equal on every key retain their input order.

Supported key column types: String, Bool, Int32, Int64, Uint32, Uint64, Float64, Float32, Timestamp. Nulls sort last.

Example: 

// Chronologically by date; break ties by value descending.
out, err := df.SortBy(
    gobi.SortKey{Column: "date"},
    gobi.SortKey{Column: "value", Descending: true},
)

func (*Frame) String 

func (f *Frame) String() string

String returns a debug representation of the frame. Not intended for pretty printing at scale.

func (*Frame) Table 

func (f *Frame) Table() arrow.Table

Table returns an arrow.Table view of the frame. The returned table shares buffers with the frame — releasing one releases the other.

func (*Frame) Tail 

func (f *Frame) Tail(n int) *Frame

Tail returns a Frame with the last n rows (default 5).

func (*Frame) Take 

func (f *Frame) Take(indexes []int) (*Frame, error)

Take returns a new Frame consisting of rows selected by the given indexes, in the order given. Duplicates are allowed. Out-of-range indexes produce an error.

func (*Frame) Union 

func (f *Frame) Union(other *Frame, cols ...string) (*Frame, error)

Union returns rows in either f or other, deduplicated over cols. When cols is nil or empty, uniqueness is determined over every column. Uses the same composite-key encoding as GroupBy so null-vs-null comparisons collide (nulls are treated as equal for set membership — pandas / polars semantics, not SQL semantics).

Both frames must share the exact same schema — see Concat for the compatibility rules and error format.

Equivalent to `f.Concat(other).Unique(cols...)` but bundled for clarity of intent.

func (*Frame) Unique 

func (f *Frame) Unique(cols ...string) (*Frame, error)

Unique returns a Frame with duplicate rows removed. Uniqueness is determined by the columns listed in cols; other columns come along for the ride from the first occurrence of each distinct key.

When cols is empty, distinctness is computed over every column (SQL `DISTINCT *`). Any listed column must be a hashable arrow type (String, LargeString, Int64, Int32, Uint64, Uint32, Float64, Bool, Timestamp) — the same set GroupBy accepts.

Semantics match pandas' `df.drop_duplicates(subset=cols)` and polars' `df.unique(subset=cols)`: the result preserves the full input schema and first-occurrence order. If you only want the distinct values of a single column, use `df.Column(name).Unique()` which returns a Series.

This is the whole-frame `distinct` primitive. For per-group distinct counts, use `GroupBy(...).Agg(Aggregation{Kind: AggNUnique})`.

func (*Frame) ValueCounts 

func (f *Frame) ValueCounts(col string) (*Frame, error)

ValueCounts returns a two-column Frame of (value, count) pairs counting the occurrences of each distinct value of col, sorted descending by count. Ties are broken by the value itself ascending — makes the output stable / deterministic even when several distinct values share a frequency.

The output columns are named `col` (matching the input) and `count` (Int64). Null values in the source column are counted as their own group under the null value slot — the count row for null contains a null in the value column and its count in the count column.

Equivalent to pandas' `Series.value_counts()` / polars' `Series.value_counts()`. For per-group value counts (e.g. count of role per department), you want a nested output that gobi doesn't have a schema for yet — build it via two-level GroupBy today or wait for list-typed columns.

func (*Frame) WithColumn 

func (f *Frame) WithColumn(name string, s Series) (*Frame, error)

WithColumn returns a new Frame with s appended (or replaced, if a column already exists with that name). The returned Frame independently ref-counts every column, so the caller can Release either Frame without affecting the other's buffers.

s.Len() must equal f.NumRows() unless f has zero columns.

The Series' arrow.Field is carried over — its Nullable flag, geometry metadata, and any other field-level attributes — with only the Name replaced by name. This preserves geometry-column identification when swapping in a WKB Binary column derived from a user function.

func (*Frame) WithColumnExpr 

func (f *Frame) WithColumnExpr(name string, e Expr) (*Frame, error)

WithColumnExpr returns a new Frame with e evaluated and attached under name. If a column named name already exists, it is replaced in place; otherwise it is appended.

Example: 

// Add a computed column: usd_price = eur_price * 1.08
out, err := df.WithColumnExpr("usd_price",
    gobi.Col("eur_price").Mul(gobi.Lit(1.08)),
)

type GeoParquetColumnMeta 

type GeoParquetColumnMeta struct {
	Encoding      string         `json:"encoding"`
	GeometryTypes []string       `json:"geometry_types"`
	CRS           map[string]any `json:"crs,omitempty"`
	Bbox          []float64      `json:"bbox,omitempty"`
}

GeoParquetColumnMeta is the per-column entry inside GeoParquetMetadata.

type GeoParquetMetadata 

type GeoParquetMetadata struct {
	Version       string                          `json:"version"`
	PrimaryColumn string                          `json:"primary_column"`
	Columns       map[string]GeoParquetColumnMeta `json:"columns"`
}

GeoParquetMetadata is the JSON payload written under the "geo" key of a GeoParquet file's Arrow-level metadata.

func BuildGeoParquetMetadata 

func BuildGeoParquetMetadata(f *Frame) (*GeoParquetMetadata, error)

BuildGeoParquetMetadata scans f and produces a GeoParquet metadata blob describing its geometry columns. Every column tagged as a geometry column (via GeometryField) is scanned once to compute its bounding box and the set of geometry types it contains.

func ParseGeoParquetMetadata 

func ParseGeoParquetMetadata(raw string) (*GeoParquetMetadata, error)

ParseGeoParquetMetadata decodes the JSON blob under the "geo" key.

type GroupBy 

type GroupBy struct {
	// contains filtered or unexported fields
}

GroupBy partitions a Frame by the values in one or more key columns. The keys must be of a hashable Arrow type (String, Int64, Int32, Bool, Float64).

func (*GroupBy) Agg 

func (g *GroupBy) Agg(aggs ...Aggregation) (*Frame, error)

Agg computes the requested aggregations over each group, returning a Frame whose first N columns are the group keys (in the order passed to GroupBy) and whose remaining columns are the aggregations in order.

When the group-by uses exactly one hashable key column with a single Arrow chunk and every aggregation column is a single-chunk primitive numeric type, the fast path (aggFast) is taken — it avoids the per-row byte-slice hashing and chunk walks that dominate the general path. All other shapes fall back to the multi-key path below.

type JoinType 

type JoinType uint8

JoinType selects the join behavior. 

const (
	// JoinInner returns rows where the key exists on both sides.
	JoinInner JoinType = iota
	// JoinLeft returns every row from the left frame, with nulls where the
	// right side has no matching key.
	JoinLeft
	// JoinRight returns every row from the right frame, with nulls where
	// the left side has no matching key. Row order follows the right
	// frame's row order.
	JoinRight
	// JoinFull (a.k.a. FULL OUTER) returns the union of JoinLeft and
	// JoinRight: every row from both frames, matched where possible and
	// null-padded on the missing side otherwise. Left rows come first
	// (in left-row order), followed by unmatched right rows in right-row
	// order.
	JoinFull
	// JoinSemi returns each left row that has at least one match on the
	// right, without duplication when there are multiple matches. Only
	// left-side columns are emitted.
	JoinSemi
	// JoinAnti returns each left row that has NO match on the right.
	// Only left-side columns are emitted. Left rows with a null key are
	// treated as unmatched.
	JoinAnti
)

type LazyFrame 

type LazyFrame struct {
	// contains filtered or unexported fields
}

LazyFrame is a Frame that hasn't been computed yet — a chain of operations expressed as a plan tree. Building a LazyFrame does no work; each fluent method appends a node to the plan. Call Collect() to actually run it.

Today (Slice 1 of Layer 2-3) Collect() dispatches each plan node to the existing eager Frame methods. There is no optimizer yet — a LazyFrame chain runs at exactly the same speed as the equivalent eager code. The value here is API shape:

  • Inspect a plan without running it: Explain() / Schema().
  • Compose complex pipelines without naming intermediate Frames.
  • Ready surface for Layer 4 optimizer passes and Layer 6 vectorized executor.

func NewLazyFrame 

func NewLazyFrame(plan LogicalPlan) *LazyFrame

NewLazyFrame constructs a LazyFrame from an arbitrary LogicalPlan. Reserved for advanced users building their own plan trees (custom scan nodes, plan-tree rewrites). Most callers should use Frame.Lazy() and the fluent builders.

func (*LazyFrame) Collect 

func (lf *LazyFrame) Collect() (*Frame, error)

Collect materializes the LazyFrame into a concrete *Frame. Runs the default optimizer, compiles the plan into a tree of ExecOperators, and drives execution via the streaming executor.

Streaming operators (Filter, Project, WithColumn, Drop, Limit) pull one batch at a time; blocking operators (Sort, Aggregate, Join, Tail) buffer their input to a Frame and delegate to the eager engine. Peak memory is bounded to one batch per streaming node plus the accumulated Frame at each blocking node.

This is where errors surface: bad expressions, type mismatches, unknown columns, scan failures.

func (*LazyFrame) CollectRaw 

func (lf *LazyFrame) CollectRaw() (*Frame, error)

CollectRaw skips both the optimizer AND the streaming executor, executing the plan tree via the bottom-up whole-Frame walker used before Layer 6. Useful for debugging optimizer bugs, for benchmarks that isolate the eager engine's cost, and as a correctness oracle for the executor's tests. Prefer Collect for real use.

func (*LazyFrame) DropColumn 

func (lf *LazyFrame) DropColumn(name string) *LazyFrame

DropColumn appends a Drop node. If the named column is missing at Collect time, the underlying Frame.DropColumn surfaces ErrColumnNotFound.

func (*LazyFrame) Explain 

func (lf *LazyFrame) Explain() string

Explain returns a human-readable representation of the plan tree. Deepest node first (which is how the tree evaluates: bottom-up). Handy for debugging what a chain of fluent methods actually built.

Example: 

Limit(1000)
  Project(col("id"), (col("value") * lit(1.08)) AS "usd")
    Filter((col("value") > lit(100)))
      Scan[frame](500 rows × 3 cols)

func (*LazyFrame) ExplainOptimized 

func (lf *LazyFrame) ExplainOptimized() string

ExplainOptimized returns the Explain output for the plan tree after the default rule set has been applied.

func (*LazyFrame) ExplainPhysical 

func (lf *LazyFrame) ExplainPhysical() string

ExplainPhysical returns a description of how the LazyFrame will be executed — which operators stream, which materialize, which scan strategy each source uses. Predicts the compilation choices Compile would make without actually building the executor tree (no goroutines started, no right-side materialization for joins, no file I/O). Cheap enough to call in tests and inspection loops.

Format: outer-most node first, one node per line, two-space indent per depth. Each label is prefixed with the strategy ("Streaming*" or "Materialize*") so the choice is visible at a glance.

Example (analytical pipeline over parquet): 

StreamingAggregate(keys=[region], aggs=[value_sum])
  StreamingJoin(inner, left.user_id = right.user_id)
    Filter((col("value") > lit(100)))
      ScanFile[stream, parquet]("events.parquet", cols=[user_id value region])
    ScanFrame(100 rows × 2 cols)

func (*LazyFrame) Filter 

func (lf *LazyFrame) Filter(cond Expr) *LazyFrame

Filter appends a Filter node. cond must produce a Boolean Series when evaluated; the error surfaces at Collect() time, not here.

func (*LazyFrame) GroupBy 

func (lf *LazyFrame) GroupBy(keys ...string) *LazyGroupBy

GroupBy returns a lazy group-by builder. Chain .Agg(aggs...) to get back a LazyFrame whose Collect() runs Frame.GroupBy(keys...).Agg(aggs...).

The two-step shape mirrors the eager API — no ambiguity about what gb, err := lf.GroupBy(...) means (there's no error return here, missing keys surface at Collect).

func (*LazyFrame) Head 

func (lf *LazyFrame) Head(n int) *LazyFrame

Head returns the first n rows. Alias for Limit — the plan node is the same. Kept as a separate method for pandas-shaped ergonomics.

func (*LazyFrame) Join 

func (lf *LazyFrame) Join(right *LazyFrame, leftKey, rightKey string, kind JoinType) *LazyFrame

Join appends a Join node combining lf and right on (leftKey, rightKey). Semantics match Frame.Join for all six JoinType values; see that method's documentation for null-key handling and key- coalescing on Right/Full joins.

func (*LazyFrame) Limit 

func (lf *LazyFrame) Limit(n int) *LazyFrame

Limit appends a Limit node that keeps the first n rows. Negative or zero n produces an empty result at Collect() time.

func (*LazyFrame) Optimize 

func (lf *LazyFrame) Optimize() LogicalPlan

Optimize returns the optimizer-rewritten plan without executing. Inspection aid: pair with String() / Explain() to see what a chain of fluent methods reduces to.

func (*LazyFrame) Plan 

func (lf *LazyFrame) Plan() LogicalPlan

Plan returns the underlying plan tree. Used by tree walkers and by tests exercising plan structure directly.

func (*LazyFrame) Schema 

func (lf *LazyFrame) Schema() *arrow.Schema

Schema returns the plan's output schema without executing anything. For plans that involve type inference (Project, WithColumn), fields whose expressions can't be statically typed are reported with a nil Type — Collect() will surface the underlying evaluation error.

func (*LazyFrame) Select 

func (lf *LazyFrame) Select(exprs ...Expr) *LazyFrame

Select appends a Project node — the resulting LazyFrame contains only the columns produced by exprs, in that order. Each expression's output column name comes from Namer.OutputName if the expression has one (Col("id") → "id", something.Alias("x") → "x"); otherwise a positional default ("expr_0", "expr_1", ...) is used.

Zero expressions is a valid but empty Select — the resulting Frame has 0 columns. Add an Alias if you want a non-default output name for computed columns.

func (*LazyFrame) SortBy 

func (lf *LazyFrame) SortBy(keys ...SortKey) *LazyFrame

SortBy appends a Sort node. Semantics match Frame.SortBy: multi-key stable, nulls-last, direction per-key.

func (*LazyFrame) Tail 

func (lf *LazyFrame) Tail(n int) *LazyFrame

Tail returns the last n rows. Unlike Head, this cannot be implemented as a plain Limit — offset depends on the total row count, which isn't known until the input has been materialized. Dispatches to Frame.Tail at Collect time.

func (*LazyFrame) WithColumn 

func (lf *LazyFrame) WithColumn(name string, e Expr) *LazyFrame

WithColumn appends a WithColumn node. If a column named name already exists at this point in the plan, it will be replaced; otherwise the new column is appended.

type LazyGroupBy 

type LazyGroupBy struct {
	// contains filtered or unexported fields
}

LazyGroupBy is the intermediate builder returned by LazyFrame.GroupBy. Its only method, Agg, closes the group-by out into a LazyFrame.

func (*LazyGroupBy) Agg 

func (lg *LazyGroupBy) Agg(aggs ...Aggregation) *LazyFrame

Agg attaches the given aggregations and returns a LazyFrame whose output schema is [group keys...] + [agg outputs...] in the order given. Naming, type mapping, and null semantics match the eager GroupBy.Agg.

type LogicalPlan 

type LogicalPlan interface {
	// Schema returns the output schema of this node. Called during
	// LazyFrame.Schema() and by nodes that need to inspect their
	// input's shape (e.g. Project computing its output columns).
	Schema() *arrow.Schema

	// Children returns the immediate sub-plans, in evaluation order
	// (empty for leaves). Used by tree walkers, optimizers, and
	// Explain().
	Children() []LogicalPlan

	// String returns a single-line description of this node —
	// no children, no formatting. Used as the label in Explain
	// output; the tree walker handles indentation and recursion.
	String() string
}

LogicalPlan represents a node in a query plan tree. Plans are immutable data structures — building a plan does not execute anything. LazyFrame consumes a LogicalPlan and executes it via Collect().

A future optimizer will walk plan trees, rewrite them, and translate them to a physical plan for execution. Slice 1 (this file) provides the tree shape and node types; Collect() dispatches each node to the existing eager engine.

func NewScanNode 

func NewScanNode(label string, schema *arrow.Schema, read func() (*Frame, error), opts ...ScanOption) LogicalPlan

NewScanNode constructs a leaf plan node representing a deferred I/O source. label is used in Explain output (e.g. "Scan[parquet](path)"), schema is the eagerly-inferred output schema (or nil if unknown at construction), and read is called by Collect() to materialize the scan's *Frame.

Optional ScanOptions declare capabilities the optimizer can exploit — WithColumnProjection enables projection pushdown from any Project/Filter above the scan.

Exposed for source packages (parquetio, csvio, ...) to build scan leaves without a per-format node type in the core package. Most callers should use the format-specific wrapper (parquetio.ScanFile and friends) instead of building nodes by hand.

func Optimize 

func Optimize(plan LogicalPlan, rules ...Rule) LogicalPlan

Optimize applies rules to plan until either nothing changes or the iteration cap is hit. Returns the (possibly rewritten) root. If rules is empty, DefaultRules() is used.

type Namer 

type Namer interface {
	OutputName() string
}

Namer is implemented by expression nodes that carry a meaningful output column name. Select and WithColumn use it to derive default column names (Col("id") → "id", Col(x).Alias("y") → "y").

Nodes that don't implement Namer, or whose OutputName is empty, fall back to a positional default ("expr_0", "expr_1", ...).

type Option 

type Option interface {
	// contains filtered or unexported methods
}

Option configures a parallel gobi operation. Values are applied to an internal options struct in the order the caller supplies them, so later options override earlier ones.

Every parallel entry point (currently SJoin; more to come) accepts a trailing `...Option`. Callers that don't care can omit them entirely.

func Workers 

func Workers(n int) Option

Workers overrides the maximum number of parallel workers used by the operation it is passed to.

  • Workers(n) with n > 0 caps parallelism at n workers.
  • Workers(0) or Workers(-1) means "use the package default" (see SetMaxParallelism); if that too is unset, GOMAXPROCS is used.
  • Workers(1) forces sequential execution.

type PredicateSink 

type PredicateSink interface {
	LogicalPlan
	// ApplyPredicate returns a plan node identical to the receiver
	// but with pred available for row-group / bloom-filter skipping
	// at read time. The Filter node above the scan is not removed
	// (row-level filtering still runs after coarse skipping), so
	// callbacks may return the receiver unchanged when they decide
	// the hint isn't useful.
	ApplyPredicate(pred Expr) LogicalPlan
}

PredicateSink is implemented by scan nodes that can accept a filter-predicate hint from the optimizer. Rules test via type assertion; scans that don't implement it are left untouched by predicate pushdown.

type ProjectableScan 

type ProjectableScan interface {
	LogicalPlan
	// ProjectColumns returns a plan node identical to the receiver
	// but restricted to the given column names. If projection is
	// not beneficial (e.g. caller already set an explicit column
	// list), implementations should return the receiver unchanged.
	ProjectColumns(cols []string) LogicalPlan
}

ProjectableScan is implemented by scan nodes that can accept a column-projection hint from the optimizer. Rules test for this interface via type assertion; scans that don't implement it are left untouched by projection pushdown.

type Resampler 

type Resampler struct {
	// contains filtered or unexported fields
}

Resampler bins rows of a Frame by fixed time intervals aligned to the Unix epoch. Use Frame.ResampleEvery to construct one, then call Agg to produce a downsampled Frame.

Bucket alignment: bucketStart = floor(unix_ns / interval_ns) * interval_ns. A 1-hour resample produces buckets that start on the hour in UTC — regardless of any timezone label on the input series. If you need calendar-aligned resampling (e.g. daily buckets that start at midnight New York time), truncate to a calendar boundary first and group on that.

func (*Resampler) Agg 

func (r *Resampler) Agg(aggs ...Aggregation) (*Frame, error)

Agg computes the requested aggregations over each non-empty bucket. The output frame has one row per bucket, ordered by bucket start, with the time column named the same as the input's time column followed by one column per aggregation. Empty buckets are omitted (this mirrors GroupBy's behavior).

type Rule 

type Rule interface {
	Name() string
	Apply(plan LogicalPlan) (rewritten LogicalPlan, changed bool)
}

Rule is a single plan-tree rewrite. A rule is a pure function from a plan to an equivalent plan, potentially cheaper to execute. The optimizer applies rules to a fixed point — no rule reports a change on the final pass.

Rules should walk the whole tree themselves; the optimizer does not traverse for them. This lets a rule choose bottom-up or top-down order depending on what it does.

func DefaultRules 

func DefaultRules() []Rule

DefaultRules returns the rule set applied by Optimize when no explicit list is passed. Order is a best-effort ordering that enables downstream folds — FoldConstants can turn a filter's condition into a literal, which RemoveTrivialTrueFilter then eliminates. The optimizer runs to a fixed point, so any correct interleaving eventually converges.

type ScanOption 

type ScanOption func(*scanFileNode)

ScanOption configures optional capabilities on a scan node. Source packages (parquetio, csvio, ...) pass these to NewScanNode to declare features the optimizer can exploit — currently just column-projection pushdown.

func WithColumnProjection 

func WithColumnProjection(fn func(cols []string) LogicalPlan) ScanOption

WithColumnProjection registers a callback that returns a new scan node projected to the given column names. The optimizer's projection-pushdown rule calls this when it determines a scan source is being asked for a subset of its columns.

The callback should:

  • Return nil if projection doesn't apply (e.g. the caller already restricted columns explicitly). Treated as "no change" — the receiver stays in the plan.
  • Otherwise return a fresh scan node with the projection applied — the optimizer replaces the old node with what this callback returns.

cols is guaranteed non-empty and sorted; entries outside the scan's schema are the callback's responsibility to filter out.

func WithParallelStreamReads 

func WithParallelStreamReads(fn func() []func(cb func(*Frame) error) error) ScanOption

WithParallelStreamReads registers a callback that returns N stream-read closures, each processing a disjoint portion of the source. The Layer 6 executor spawns one producer goroutine per closure and fan-ins their batches into the downstream pipeline.

The callback is responsible for deciding N (typically bounded by GOMAXPROCS and by the source's natural partition count — e.g. number of parquet row-groups). Return nil (or a slice of length 0-1) to signal "no parallel version available"; the executor falls back to WithStreamRead in that case.

Semantics: batches from different workers arrive in unspecified order. Downstream operators that require order (Sort, Tail) buffer and reorder internally, so this is safe to enable for any plan shape.

func WithPredicatePushdown 

func WithPredicatePushdown(fn func(pred Expr) LogicalPlan) ScanOption

WithPredicatePushdown registers a callback that returns a new scan node with a predicate baked in. The optimizer's PushPredicateToScan rule calls this when a Filter sits directly above the scan; the scan source is expected to use the predicate for row-group / bloom-filter skipping at read time.

The Filter node above the scan is NOT removed after pushdown — row-group skipping is coarse (whole groups only), so the row-level Filter still runs for correctness. Callbacks should treat the predicate as a hint, not a guarantee.

nil return = "no change" (same convention as WithColumnProjection).

func WithStreamRead 

func WithStreamRead(fn func(cb func(*Frame) error) error) ScanOption

WithStreamRead registers a streaming callback API for scan sources that support incremental read (parquetio.ReadFileChunksFunc, csvio.ReadFileChunksFunc). fn should call cb once per record batch and honor the batch-lifetime contract of the source package (typically: batches are Released after cb returns).

When present, the Layer 6 executor uses this callback for true bounded-memory streaming through the plan. When absent, the executor falls back to the `read` closure (whole-file materialize then batch — correct but not memory-bounded).

type Series 

type Series struct {
	// contains filtered or unexported fields
}

Series is a single named, typed Arrow column.

A Series does not own its column: constructors that receive an *arrow.Column treat it as borrowed. Callers wanting shared lifetime should Retain the underlying column; DataFrame.Release does not release Series contents.

func NewSeries 

func NewSeries(col *arrow.Column) Series

NewSeries returns a Series wrapping col. The Series takes the column's name from col.Name(); its field is derived from col.Field().

func NewTimestampSeries 

func NewTimestampSeries(name string, ts []time.Time, validity []bool) Series

NewTimestampSeries builds a Timestamp[ns] Series from the given time.Time values. Passing a nil validity slice means every row is valid; otherwise validity[i]==false marks row i as null (its value is ignored).

func PointsFromXY 

func PointsFromXY(x, y Series, crs int32) (Series, error)

PointsFromXY builds a geometry Series of 2D WKB Points from two coordinate columns. x and y must be numeric (Float64, Float32, Int64, or Int32) and the same length. Mixed-type inputs are promoted to Float64. Null values on either side emit a null geometry for that row.

The returned Series is a WKB Binary column tagged with geometry metadata + the given EPSG code, so it plugs directly into Frame.WithColumn, Frame.SJoin, GeoParquet write paths, and other geometry-aware operations.

Modeled on geopandas.points_from_xy — the intended flow is to build a geometry column from two attribute columns without hand-rolling the WKB encoding: 

lat, _ := df.Column("lat")
lng, _ := df.Column("lng")
geom, _ := gobi.PointsFromXY(lng, lat, 4326)   // x=lng, y=lat
df, _ = df.WithColumn("geometry", geom)

Note the argument order: x first, y second. In geographic coordinates that means longitude first, latitude second — matching GeoJSON / WKB / shapefile conventions (and geopandas).

func PointsFromXYZ 

func PointsFromXYZ(x, y, z Series, crs int32) (Series, error)

PointsFromXYZ is the 3D variant of PointsFromXY. z must be numeric and the same length as x and y; rows with a null z produce null geometries even if x and y are valid.

The resulting Point geometries carry HasZ=true so downstream WKB encoding emits XYZ type codes (1001..) rather than 2D (1..).

func (Series) Add 

func (s Series) Add(o Series) (Series, error)

Add returns s + o element-wise.

func (Series) AddDuration 

func (s Series) AddDuration(d time.Duration) (Series, error)

AddDuration returns a new Timestamp[ns] Series with d added to each row. Nulls propagate.

func (Series) AddScalar 

func (s Series) AddScalar(v float64) (Series, error)

AddScalar returns s + v element-wise (result is float64).

func (Series) Column 

func (s Series) Column() *arrow.Column

Column returns the underlying Arrow column. Do not mutate the returned value; use Series methods for slicing.

func (Series) Concat 

func (s Series) Concat(others ...Series) (Series, error)

Concat returns a new Series with s's values followed by each other's values, in order. All series must share the same arrow type; type mismatches return an error naming both sides.

The result is a multi-chunk column — each input becomes a chunk. Zero-copy: no memcpy of the underlying arrow buffers.

func (Series) Count 

func (s Series) Count() int

Count returns the number of non-null rows.

func (Series) DataType 

func (s Series) DataType() arrow.DataType

DataType returns the Arrow data type.

func (Series) Day 

func (s Series) Day() (Series, error)

Day returns each row's day-of-month [1..31] as Int64.

func (Series) DayOfYear 

func (s Series) DayOfYear() (Series, error)

DayOfYear returns each row's day-of-year [1..366] as Int64.

func (Series) Diff 

func (s Series) Diff(n int) (Series, error)

Diff returns a Series holding the element-wise difference between s and its n-lagged self: `out[i] = s[i] - s[i-n]`. The first n positions are null (no previous value to subtract). n must be positive; n=0 is disallowed (would trivially be all zeros); use Shift(-n) if you want a negative-lag first-difference.

Only numeric types are supported (matching the underlying Series.Sub). Non-numeric callers should compose Shift + a domain-specific comparison instead.

This is the columnar equivalent of pandas / polars `diff(n)` used for period-over-period changes, discrete derivatives, and change-point detection.

func (Series) DiffDuration 

func (s Series) DiffDuration(other Series) (Series, error)

DiffDuration returns an Int64 Series whose row i is (s[i] - other[i]) expressed in nanoseconds. Nulls on either side produce nulls in the output.

func (Series) Difference 

func (s Series) Difference(other Series) (Series, error)

Difference returns the distinct non-null values in s that are absent from other, in first-occurrence order. Types must match.

func (Series) Div 

func (s Series) Div(o Series) (Series, error)

Div returns s / o element-wise, promoting to float64. Division by zero yields ±Inf or NaN per IEEE 754.

func (Series) DivScalar 

func (s Series) DivScalar(v float64) (Series, error)

DivScalar returns s / v element-wise.

func (Series) Eq 

func (s Series) Eq(o Series) (Series, error)

Eq returns a boolean Series that is true where s[i] == o[i].

func (Series) EqScalar 

func (s Series) EqScalar(v float64) (Series, error)

EqScalar returns a boolean Series true where s[i] == v.

func (Series) EqTime 

func (s Series) EqTime(t time.Time) (Series, error)

EqTime returns a Boolean Series true where s[i] == t.

func (Series) Ge 

func (s Series) Ge(o Series) (Series, error)

Ge returns a boolean Series that is true where s[i] >= o[i].

func (Series) GeTime 

func (s Series) GeTime(t time.Time) (Series, error)

GeTime returns a Boolean Series true where s[i] >= t.

func (Series) GeomArea 

func (s Series) GeomArea(u geometry.Unit) (Series, error)

GeomArea returns a Float64 Series holding the planar (XY) area of each geometry in s, in u². Non-polygonal geometries contribute 0. Null geometries produce null values.

func (Series) GeomBounds 

func (s Series) GeomBounds() (*Frame, error)

GeomBounds returns a Frame with four Float64 columns — MinX, MinY, MaxX, MaxY — one row per input geometry. Null geometries produce four nulls.

func (Series) GeomCentroid 

func (s Series) GeomCentroid() (Series, error)

GeomCentroid returns a geometry Series holding the centroid of each input geometry as a Point (encoded as WKB). Null inputs produce nulls. The output column inherits s's CRS.

func (Series) GeomLength 

func (s Series) GeomLength(u geometry.Unit) (Series, error)

GeomLength returns a Float64 Series holding the planar (XY) length of each geometry in u. Non-linear geometries contribute 0. Null geometries produce null values.

func (Series) Geometry 

func (s Series) Geometry(i int) (geometry.Geometry, error)

Geometry decodes the WKB at row i into a Geometry. Returns ErrNotGeometry if the series is not a geometry column.

func (Series) Gt 

func (s Series) Gt(o Series) (Series, error)

Gt returns a boolean Series that is true where s[i] > o[i].

func (Series) GtScalar 

func (s Series) GtScalar(v float64) (Series, error)

GtScalar returns a boolean Series true where s[i] > v.

func (Series) GtTime 

func (s Series) GtTime(t time.Time) (Series, error)

GtTime returns a Boolean Series true where s[i] > t.

func (Series) Head 

func (s Series) Head(n int) Series

Head returns a Series with the first n rows. n<=0 means default 5.

func (Series) Hour 

func (s Series) Hour() (Series, error)

Hour returns each row's hour-of-day [0..23] as Int64 (UTC).

func (Series) Intersect 

func (s Series) Intersect(other Series) (Series, error)

Intersect returns the distinct non-null values present in both s and other, in first-occurrence order. Types must match.

func (Series) IsDateTime 

func (s Series) IsDateTime() bool

IsDateTime reports whether s is a Timestamp / Date32 / Date64 column.

func (Series) IsGeometry 

func (s Series) IsGeometry() bool

IsGeometry reports whether the series is tagged as a WKB geometry column.

func (Series) Le 

func (s Series) Le(o Series) (Series, error)

Le returns a boolean Series that is true where s[i] <= o[i].

func (Series) LeTime 

func (s Series) LeTime(t time.Time) (Series, error)

LeTime returns a Boolean Series true where s[i] <= t.

func (Series) Len 

func (s Series) Len() int

Len returns the number of rows.

func (Series) Lt 

func (s Series) Lt(o Series) (Series, error)

Lt returns a boolean Series that is true where s[i] < o[i].

func (Series) LtScalar 

func (s Series) LtScalar(v float64) (Series, error)

LtScalar returns a boolean Series true where s[i] < v.

func (Series) LtTime 

func (s Series) LtTime(t time.Time) (Series, error)

LtTime returns a Boolean Series true where s[i] < t.

func (Series) Max 

func (s Series) Max() (float64, error)

Max returns the maximum non-null numeric value, or NaN when the series has no non-null rows.

func (Series) Mean 

func (s Series) Mean() (float64, error)

Mean returns the arithmetic mean of non-null numeric values, or NaN when the series has no non-null rows.

func (Series) Min 

func (s Series) Min() (float64, error)

Min returns the minimum non-null numeric value, or NaN when the series has no non-null rows.

func (Series) Minute 

func (s Series) Minute() (Series, error)

Minute returns each row's minute-of-hour [0..59] as Int64.

func (Series) Month 

func (s Series) Month() (Series, error)

Month returns each row's calendar month [1..12] as Int64.

func (Series) Mul 

func (s Series) Mul(o Series) (Series, error)

Mul returns s * o element-wise.

func (Series) MulScalar 

func (s Series) MulScalar(v float64) (Series, error)

MulScalar returns s * v element-wise.

func (Series) NUnique 

func (s Series) NUnique() (int64, error)

NUnique returns the count of distinct non-null values in s.

Equivalent to `s.Unique().Len()` but skips the arrow array construction. Uses the same byte-encoding as GroupBy so numeric bit-equal values collapse identically. O(n) time, O(distinct) memory.

func (Series) Name 

func (s Series) Name() string

Name returns the column name.

func (Series) Ne 

func (s Series) Ne(o Series) (Series, error)

Ne returns a boolean Series that is true where s[i] != o[i].

func (Series) NeTime 

func (s Series) NeTime(t time.Time) (Series, error)

NeTime returns a Boolean Series true where s[i] != t.

func (Series) RollingCount 

func (s Series) RollingCount(window int) (Series, error)

RollingCount returns an Int64 Series holding the count of non-null values in each window. Rows before the first full window are null.

func (Series) RollingMax 

func (s Series) RollingMax(window int) (Series, error)

RollingMax returns a Float64 Series holding the maximum non-null value over each window.

func (Series) RollingMean 

func (s Series) RollingMean(window int) (Series, error)

RollingMean returns a Float64 Series holding the arithmetic mean of s over each window (non-null values only).

func (Series) RollingMin 

func (s Series) RollingMin(window int) (Series, error)

RollingMin returns a Float64 Series holding the minimum non-null value over each window.

func (Series) RollingSum 

func (s Series) RollingSum(window int) (Series, error)

RollingSum returns a Float64 Series where row i is the sum of s over the closed window [i-window+1 .. i]. window must be >= 1.

func (Series) Row 

func (s Series) Row(i int) (Series, error)

Row returns a Series containing the single row at index i.

func (Series) Second 

func (s Series) Second() (Series, error)

Second returns each row's second-of-minute [0..59] as Int64.

func (Series) Shift 

func (s Series) Shift(n int) (Series, error)

Shift returns a new Series with values shifted by n positions. Positive n shifts values forward (down): the first n positions become null, and each output position i gets the source value from position i-n when that index is in bounds.

Negative n shifts backward (up): the last |n| positions become null; output position i gets the source value from position i+|n|.

n=0 is a no-op that still returns a fresh Series (new arrow array, same values). Extreme |n| >= Len() returns an all-null Series of the same type + name — same as pandas / polars.

Nulls in the source at the source index propagate to the output. The output preserves the source's arrow type, name, and field metadata (including the geometry tag, if any).

func (Series) Sub 

func (s Series) Sub(o Series) (Series, error)

Sub returns s - o element-wise.

func (Series) SubDuration 

func (s Series) SubDuration(d time.Duration) (Series, error)

SubDuration returns a new Timestamp[ns] Series with d subtracted from each row.

func (Series) SubScalar 

func (s Series) SubScalar(v float64) (Series, error)

SubScalar returns s - v element-wise.

func (Series) Sum 

func (s Series) Sum() (float64, error)

Sum returns the sum of non-null numeric values.

func (Series) Tail 

func (s Series) Tail(n int) Series

Tail returns a Series with the last n rows. n<=0 means default 5.

func (Series) TimeAt 

func (s Series) TimeAt(i int) (time.Time, bool, error)

TimeAt returns the value at row i as a time.Time, plus a validity flag (false for null). The returned time.Time carries the series' timezone (UTC if the series is tz-naive). Errors for non-datetime series.

func (Series) Timezone 

func (s Series) Timezone() string

Timezone returns the IANA timezone label on s (e.g. "America/New_York"), or "" if the series is tz-naive or not a Timestamp column.

func (Series) TruncateTo 

func (s Series) TruncateTo(unit TimeUnit) (Series, error)

TruncateTo returns a new Timestamp[ns] Series with each row rounded down to the nearest boundary of the given unit. UnitDay truncates to 00:00:00 UTC of the same date.

func (Series) TruncateToCalendar 

func (s Series) TruncateToCalendar(unit CalendarUnit) (Series, error)

TruncateToCalendar truncates each row to a calendar boundary (week, month, year). If s carries a timezone, the boundary is computed in that timezone (so "start of month" is midnight local time in the tz, not UTC). Week uses ISO semantics: previous Monday at 00:00.

func (Series) Union 

func (s Series) Union(other Series) (Series, error)

Union returns the distinct non-null values in either s or other, in first-occurrence order. Types must match. Nulls are treated as equal for set membership — matching Frame.Union — but the null value itself is not emitted in the output (parity with Series.Unique which drops nulls).

func (Series) Unique 

func (s Series) Unique() (Series, error)

Unique returns a Series of distinct non-null values in first- occurrence order. The result's arrow type matches the input's.

Nulls are dropped (matches pandas / polars `unique` semantics). Callers that want a "did I see any null?" signal should combine with `Series.NullCount()`.

func (Series) Weekday 

func (s Series) Weekday() (Series, error)

Weekday returns each row's day-of-week as Int64 with Sunday=0..Saturday=6 (matches Go's time.Weekday int values).

func (Series) WithTimezone 

func (s Series) WithTimezone(tz string) (Series, error)

WithTimezone returns a copy of s that carries the given IANA timezone label. The underlying Unix-nanosecond values are NOT shifted — only the display timezone changes. This matches pandas' `tz_convert` when the input is already tz-aware, or "assume the values are absolute instants and now render them in tz" when the input is naive.

Passing tz == "" strips any existing label (result is tz-naive). Passing a name that time.LoadLocation cannot resolve returns an error.

func (Series) Year 

func (s Series) Year() (Series, error)

Year returns each row's calendar year as an Int64 Series (UTC).

type SortKey 

type SortKey struct {
	Column     string
	Descending bool
}

SortKey names a column to sort by and its direction. Compose multiple SortKeys in a single SortBy call for lexicographic (a-then-b-then-c) ordering — earlier keys have priority; later keys break ties.

Nulls sort last regardless of Descending, matching pandas / polars default behavior. NaN floats also sort last (numpy semantics).

type SpatialPredicate 

type SpatialPredicate uint8

SpatialPredicate names a binary spatial predicate for SJoin. 

const (
	// SPIntersects matches when the left and right geometries share any point.
	SPIntersects SpatialPredicate = iota
	// SPContains matches when the left geometry fully contains the right.
	SPContains
	// SPWithin matches when the left geometry lies fully within the right.
	SPWithin
)

func (SpatialPredicate) String 

func (p SpatialPredicate) String() string

type Stats 

type Stats interface {
	// MinMax returns the inclusive bounds for col over the range
	// this Stats describes. ok=false when statistics are missing.
	MinMax(col string) (minV, maxV any, ok bool)
	// NullCount returns the number of null values in col over the
	// range. ok=false when null counts weren't recorded.
	NullCount(col string) (n int64, ok bool)
	// TotalRows is the total row count of the range.
	TotalRows() int64
}

Stats reports column-level bounds used by CanPossiblyMatch to prove predicates unsatisfiable over a data range (typically a parquet row-group). Implementations are supplied by source packages: parquetio, csvio, etc.

The values returned by MinMax must be Go-typed scalars matching the column's arrow type: int64 for INT64, float64 for FLOAT64, string for STRING, bool for BOOL, and so on. Mixed types or unknown columns should signal ok=false so the caller falls back to the conservative "possibly matches" answer.

type TimeRolling 

type TimeRolling struct {
	// contains filtered or unexported fields
}

TimeRolling represents a right-anchored, time-based rolling window over a Frame. Use Frame.RollingBy to construct one.

func (*TimeRolling) Agg 

func (r *TimeRolling) Agg(column string, kind AggKind) (Series, error)

Agg computes a single aggregation over each row's trailing window and returns a Series of the aggregated values. Rows whose window is empty (e.g. all-null time or empty column) produce nulls.

type TimeUnit 

type TimeUnit uint8

TimeUnit selects a truncation granularity for Series.TruncateTo. Values intentionally include only sub-day units; week / month / year truncations use TruncateToCalendar because they require calendar (not clock) arithmetic. 

const (
	UnitNanosecond TimeUnit = iota
	UnitMicrosecond
	UnitMillisecond
	UnitSecond
	UnitMinute
	UnitHour
	UnitDay
)

Source Files

View all Source files

Directories

Path Synopsis
contrib
athenaio module
Package csvio reads and writes CSV data as gobi Frames.
 Package csvio reads and writes CSV data as gobi Frames.
Package geojsonio reads and writes gobi Frames as GeoJSON per RFC 7946.
 Package geojsonio reads and writes gobi Frames as GeoJSON per RFC 7946.
Package geometry provides 2D geometry primitives (Point, LineString, Polygon, MultiPoint) with WKB and WKT encoding, a coordinate reference system model, and common spatial operations (area, distance, centroid, convex hull, intersection tests).
 Package geometry provides 2D geometry primitives (Point, LineString, Polygon, MultiPoint) with WKB and WKT encoding, a coordinate reference system model, and common spatial operations (area, distance, centroid, convex hull, intersection tests).
Package gpkgio reads and writes gobi Frames as OGC GeoPackage (SQLite) files.
 Package gpkgio reads and writes gobi Frames as OGC GeoPackage (SQLite) files.
Package kmlio reads and writes KML (Keyhole Markup Language, OGC 12-007r2) as gobi Frames.
 Package kmlio reads and writes KML (Keyhole Markup Language, OGC 12-007r2) as gobi Frames.
Package parquetio reads and writes gobi Frames as Apache Parquet.
 Package parquetio reads and writes gobi Frames as Apache Parquet.
Package pgio reads and writes gobi Frames against PostgreSQL / PostGIS databases.
 Package pgio reads and writes gobi Frames against PostgreSQL / PostGIS databases.
Package shpio reads and writes ESRI Shapefiles as gobi Frames.
 Package shpio reads and writes ESRI Shapefiles as gobi Frames.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.