ZenQL, Powerful, Fluent Queries In Golang

Zenql — Tue, 02 Jun 2026 10:57:09 +0000

a new way to interact with your data in golang

Zenql is an internal DSL (Domain Specific Language) for golang. trying to provide a new way to work with various data-sources in an integrated environment. datasources can be of any kind, such as golang internal data-types (channels, slices) and async datasources like (csv,json,postgres db, mysql db).

zenql is inspired by C# linq and Java streams. for collection processing we have a dedicated query engine named thor, which uses operator fusion pattern to process the query-chain in a single execution unit. this approach can be very performant. our benchmarks shows that thor engine filtered a slice of 50,000,000 items in just 2 seconds.

an expressive query on collections using zenql:

Group[bool, ComplexObjectToSearch](
    From(items).Where(func(search ComplexObjectToSearch) bool {
        return search.Age > 20
    }),
    func(item ComplexObjectToSearch) bool {
        return item.Flag
    }
).Collect()

Polymorphic queries

but what about async datasources? After all the idea is that we provide A Polymorphic Query Language. the key to process async data-sources is to use the zenql stream api's. besides when we have a large data-set is that really practical to load all the data inside the memory and process them? and how can we query our data in different places and infrastructures? sometimes the data can be in-memory, other times its in a files (csv, json) and most of times its in a relational database.

luckily we provided a very capable Domain-Specific-Language for golang to be compatible with all these environments in an integrated way. the idea is that we use different kinds of adapters and zenql processing the data in a unified way.

1 - FromData to initiate stream from slices.
2 - FromChannel to initiate stream Channels.
3 - FromSqlRows to initiate stream an RDBMS (mysql,postgres).
4 - FromJsonArr to initiate stream from a json file.
5 - FromCsvFile to initiate stream from a csv file.

after initiating the stream, we can use zenql stream Pipelines to process your data.

A Real‑World Example of MySql Streams

Imagine a scenario with a large user base where you need to process users individually, such as validating each one against an external web service. Loading all records into memory is neither efficient nor scalable. Conversely, repeatedly opening and closing a database connection for every single row creates a significant performance bottleneck. With the new Zenql Streams API, you can initiate a stream using a single database connection to process rows iteratively, just like a cursor. This approach significantly reduces memory consumption and optimizes performance by eliminating unnecessary database round-trips.


    ctx, cancel := context.WithCancel(context.Background())

    defer cancel()

    constr := "root:1245Sa@tcp(127.0.0.1:30306)/Test?parseTime=true&charset=utf8mb4"

    if conn, err := db.connect("mysql",constr); err != nil {
        t.Fatal(err)
    } else {

        defer conn.Close()

        id := 0
        stream :=
            db.FromSqlRows[UserModel](ctx, conn,
                "select * from Test.users where id>?", func(rows *sql.Rows) (UserModel, error) {
                    var id, age int
                    var name string
                    var err error
                    var active bool

                    err = rows.Scan(&id, &name, &age,&active)
                    model := UserModel{
                        UserId:   id,
                        Age:      age,
                        UserName: name,
                        Active: active
                    }
                    return model, err
                }, id)

        if stream.Initiated {
            for v := range stream.FilterStream(func(model UserModel) bool {
                return model.Age > 25
            }).Throttle(time.Millisecond * 5000).Channel {

                /// business logic

                business_logic_satisfied := true

                if business_logic_satisfied {

                    result := Exec(conn, "update Test.users set Active = ? where Id =?", 1, v.UserId)
                    if result.Err != nil {
                        t.Error(result.Err)
                    } else {
                        fmt.Println(v, " - updated. ", result.RowsAffected)
                    }
                }

            }
        } else {
            fmt.Println("stream not initiated")
        }

want to try ZenQL? visit this Repository and be surprised!

DEV Community: Zenql

ZenQL, Powerful, Fluent Queries In Golang