POC: Golang implementation

[turbolytics]

We originally created sql-flow to tap into the python ML ecosystem. The Python bindings for duckdb, kafka and arrow make it really easy and ergonomic to build SQLFlow.

The goal is to see how development in go compares to that in python. The current implementation has a number of concurrent features that Go may be better suited for.

We have future work planned for enhanced ML functionality in SQLflow, so python is still an extremely compelling implementation.

refs #123

[turbolytics]

Many of the managed tables require concurrency to maintain state of the buffered tables. The thought is that managing this concurrency will be easier and safer in go than in python.

[loicalleyne]

Are you still working on this? This sounds really interesting.
In case it's useful, here's a link to an Arrow blog post about streaming inserts into DuckDB with Go.

[turbolytics]

I'd like to still explore the POC. I ran some benchmarks on the first go POC and they were really poor compared to python, so I think I was doing something really wrong with the Arrow API.

Go is still really compelling for its compiled binary, potentially opening up IoT applications , where SQLFlow runs on IoT and MQTT.

Go also seems really nice for for its concurrency support. Timeouts and flushing using python threads is already quite cumbersome.

on the python side, python ML ecosystem is really compelling.

[loicalleyne]

I see you are using marcboeker/go-duckdb, in my quacfka project I found apache/arrow-adbc/go/adbc to be faster, I was aiming for the absolute fastest insertion speeds possible with a friendly API for deeply nested data types. That being said, in the grand scheme of things as long as Arrow is the in-memory format this piece shouldn't be the bottleneck.

Fan-out/fan-in for consuming Kafka messages and then converting to Arrow records works very well in Go using channels and worker pools. Converting message data to Arrow can be either relatively simple/challenging depending on the message format, schema complexity, and API availability; arrow-go/array has a JSONReader, arrow-go/util has ProtobufMessageReflection, etc...

I found that tuning the Arrow record sizes (122880 or greater) to match the DuckDB row group size has the biggest impact on insertion speed. Doing concurrent insertions and reads in the same process and on the same database file will also slow down the insertion rate, due to this in my project I chose to implement rotating databases and to run my aggregation and COPY ... TO parquet queries in a separate service.
DuckDB isn't able to do the same amount of compression with its in-memory format as it does with its file format, if you're working only with :memory: you need to be careful with batch sizes to avoid OOM.

Unfortunately iceberg-go is not yet at feature parity with pyiceberg (partitioned writes in particular is missing), perhaps writes to Iceberg could be done by sending Arrow IPC or Arrow Flight SQL to a python service which would then write with pyiceberg.

Happy to collaborate further if you keep working on this.

[turbolytics]

Awesome, thank you! I think you got me unblocked! switched to adbc, also consulted:

https://github.com/loicalleyne/couac/blob/master/couac.go

For how to bind arrow record to duckdb 😁 🚀

[aaronS7]

Did you put this on hold?

[turbolytics]

The POC is running! I followed the approach that @loicalleyne took in their repo:

https://github.com/loicalleyne/couac

---

I'm personally a bit split on this. Right now, SQLFlow isn't getting a huge business driver to push it into go. For example, if IoT became really interested in SQLFlow I think that might be a strong argument for a single binary distribution of Go.

The python ML ecosystem is lightyears ahead of GO, so if the project decides to pursue more ML offerings then python is better suited:

Feature / Capability	Python Implementation	Go Implementation
Performance (Raw Throughput)	Medium – great for I/O bound and some CPU bound	High – excellent for CPU bound and concurrent tasks
Concurrency	Limited (GIL-bound, requires multiprocessing)	Native via goroutines – highly efficient concurrency
Deployment	Requires Python runtime and dependencies	Compiles to a single binary – ideal for Docker/IOT
Ecosystem Integration	Strong ML/DS ecosystem (NumPy, pandas, sklearn)	Strong systems/dev ecosystem (networking, infra)
ML / Data Science	Excellent – integrates seamlessly with ML toolkits	Poor – ML support is minimal or via wrappers
Library Maturity for SQL/Arrow	Mature (DuckDB, PyArrow)	Improving (Arrow C bindings, DuckDB C/Go wrappers)
Error Handling	Easier to prototype but more runtime errors	Safer and more explicit (compile-time guarantees)
Ease of Development	Fast prototyping, very expressive	More verbose but statically typed and robust
Cold Start Time	Slower (interpreter startup, imports)	Fast – sub-second binary execution
Footprint (Memory/Disk)	Medium to High – Python + dependencies	Low – single binary, good for embedded use
Cross-platform Support	Good – with some packaging effort (e.g., PyInstaller)	Excellent – cross-compilation built in
Observability/Monitoring	Rich ecosystem but less built-in	First-class tooling (pprof, Prometheus native)
Parallelism Control	Coarse – via multiprocessing/thread pools	Fine-grained with goroutines and channels
Hot Reload / Debugging	Excellent (REPL, pdb, notebooks)	Slower iteration but better for stability
Use in Embedded / Edge Devices	Rare	Common – ideal for CLI, embedded and edge systems
Learning Curve	Lower – especially for data engineers	Higher – but leads to more robust systems

Additionally, like iceberg support, python's iceberg library is more mature than go's (it can write 😁 ), but at some point duckdb will have full iceberg support. Also iceberg may be eclisped by ducklake in the future as well, so lots of things are still in flux, at this moment in time, i don't see a ton of concrete drivers pushing into go (but am totally open to them appearing 😁 )

[aaronS7]

Makes a lot of sense. Since this is a stream processing, does the integration with lots of these ML/DS ecosystems worthwhile? I ask genuinely, I don't know the right answer. Whatever route you choose, I'd be interesting in learning how to use it or potentially contributing if there is room here.

[loicalleyne]

FWIW I work in adtech where the volume of streaming data is rather large and having a simple, easy to deploy alternative to Flink would be really interesting.

[turbolytics]

@loicalleyne

Which stream processing features are essential for your use case? Which features would you consider table stakes for a stream processing solution?

[loicalleyne]

@turbolytics
The use case I'm currently investigating is to enable our product teams to send analytics data to a data lakehouse in a way that's simple and easy to deploy in CI.
MVP requirements:
Team produces in protobuf format to Kafka topic, service consumes the topic and writes the data to Iceberg table partitions without performing any enrichment or aggregations.
Topic schema would come either from a schema registry or be defined in a proto file.
Iceberg table definition would be defined in the service configuration.

For the MVP aggregations and enrichment can be done in the lakehouse, and in some cases the topic would only contain dimension data anyway. Stream processing beyond that would be a value-add for later.

To avoid writing many small files I'm thinking either a single node that's performant enough to be the only writer or several instances writing to a common cache where the data is batched together on a threshold before files are written to Iceberg.

[loicalleyne]

@turbolytics
The use case I'm currently investigating is to enable our product teams to send analytics data to a data lakehouse in a way that's simple and easy to deploy in CI.
MVP requirements:
Team produces in protobuf format to Kafka topic, service consumes the topic and writes the data to Iceberg table partitions without performing any enrichment or aggregations.
Topic schema would come either from a schema registry or be defined in a proto file.
Iceberg table definition would be defined in the service configuration.

For the MVP aggregations and enrichment can be done in the lakehouse, and in some cases the topic would only contain dimension data anyway. Stream processing would be a value-add for later.

To avoid writing many small files I'm thinking either a single node that's performant enough to be the only writer or several instances writing to a common cache where the data is batched together on a threshold before files are written to Iceberg.