perf: optimize streaming

Author: asbisen

src/params.jl

@@ -202,6 +202,18 @@ function _bind_named(sql::String, named_params::Base.Pairs)::Tuple{String, Vecto
     end
 
     sql_out = String(take!(io))
+
+    # Detect kwargs provided by the caller but never referenced in the SQL.
+    # This catches typos in either the kwarg name or the :placeholder.
+    consumed = Set(order)
+    unused = [k for k in keys(named_params) if string(k) ∉ consumed]
+    if !isempty(unused)
+        throw(QueryError(
+            "Named parameter(s) provided but not referenced in SQL: $(join(sort(string.(unused)), ", "))",
+            sql, Dict(named_params), nothing
+        ))
+    end
+
     @debug "Bound named params" order=order values=values
     return (sql_out, values)
 end

src/query.jl

@@ -224,31 +224,39 @@ end
     stream(ctx, sql; batch_size=10_000) → Channel{DataFrame}
 
 Execute `sql` and return a `Channel` that emits successive `DataFrame` batches.
-Each batch has at most `batch_size` rows.  The channel is closed automatically
-when all rows have been consumed or when an error occurs.
+The channel is closed automatically when all rows have been consumed or when an
+error occurs.
+
+The SQL is executed once and results are consumed via DuckDB's native chunk
+iterator, so streaming is O(N) regardless of result size.
+
+Each batch contains at least `batch_size` rows (it may be slightly larger
+because whole DuckDB-internal chunks of ~2048 rows are accumulated before
+emitting).
 
 ```julia
 for batch in stream(ctx, "SELECT * FROM huge_table"; batch_size=5_000)
     process(batch)
 end
 ```
-
-!!! note
-    The SQL is wrapped in a sub-query with `LIMIT … OFFSET …` which requires the
-    query engine to re-execute the underlying plan for each batch.  For best
-    performance on very large scans, consider using DuckDB's native export or
-    `COPY` instead.
 """
 function stream(ctx::QueryContext, sql::String; batch_size::Int=10_000)::Channel{DataFrame}
     Channel{DataFrame}(2) do ch
-        offset = 0
-        while true
-            batch_sql = "SELECT * FROM ($sql) AS __stream_q__ LIMIT $batch_size OFFSET $offset"
-            batch = execute(ctx, batch_sql)
-            nrow(batch) == 0 && break
-            put!(ch, batch)
-            offset += batch_size
-            nrow(batch) < batch_size && break
+        _with_conn(ctx) do conn
+            result       = DuckDB.execute(conn, sql)
+            pending      = DataFrame[]
+            pending_rows = 0
+            for chunk in Tables.partitions(result)
+                chunk_df = DataFrame(chunk)
+                push!(pending, chunk_df)
+                pending_rows += nrow(chunk_df)
+                if pending_rows >= batch_size
+                    put!(ch, vcat(pending...))
+                    empty!(pending)
+                    pending_rows = 0
+                end
+            end
+            pending_rows > 0 && put!(ch, vcat(pending...))
         end
     end
 end

test/runtests.jl

@@ -150,6 +150,17 @@ global_logger(ConsoleLogger(stderr, Logging.Warn))
         close!(ctx)
     end
 
+    @testset "Unused named param error" begin
+        ctx = QueryContext()
+        # Typo in kwarg name — :status is in SQL but :stauts is passed.
+        @test_throws QueryError execute(ctx, "SELECT :status AS s"; stauts="shipped")
+        # Extra kwarg alongside a valid one.
+        @test_throws QueryError execute(ctx, "SELECT :x AS x"; x=1, typo=2)
+        # Repeated placeholder — same param used twice; should still bind cleanly.
+        @test_nowarn execute(ctx, "SELECT :x AS a, :x AS b"; x=7)
+        close!(ctx)
+    end
+
     # ── Batch execution ────────────────────────────────────────────────────────
     @testset "execute with Vector{String}" begin
         ctx = QueryContext()
@@ -211,9 +222,12 @@ global_logger(ConsoleLogger(stderr, Logging.Warn))
         execute!(ctx, "CREATE TABLE big AS SELECT generate_series AS n FROM generate_series(1, 100)")
 
         batches = collect(stream(ctx, "SELECT * FROM big ORDER BY n"; batch_size=30))
-        @test length(batches) == 4    # 30+30+30+10
+        # Native chunk iteration: batch boundaries align to DuckDB's internal
+        # vector size (~2048), so 100 rows arrive in a single chunk and a single
+        # batch.  Assert totals and ordering rather than exact batch count.
         @test sum(nrow, batches) == 100
-        @test batches[1][1, :n] == 1
+        @test vcat(batches...)[1, :n] == 1
+        @test vcat(batches...)[end, :n] == 100
         close!(ctx)
     end