diff --git a/Dockerfile-debug b/Dockerfile-debug index 9179856..7818991 100644 --- a/Dockerfile-debug +++ b/Dockerfile-debug @@ -1,6 +1,13 @@ FROM ubuntu:latest -ARG UID +ARG UID ARG GID + +# LAGraph source to build. Defaults reproduce the pinned release; override to +# try a branch/PR, e.g. LAGRAPH_REF=matrix-sum for GraphBLAS/LAGraph#407. +ARG LAGRAPH_REPO=https://github.com/GraphBLAS/LAGraph.git +ARG LAGRAPH_REF=v1.2.1 +# Build the experimental matrix_sum aggregate (needs the matrix-sum LAGraph branch). +ARG WITH_MATRIX_SUM=0 # install base dependences RUN apt-get update && \ @@ -71,12 +78,17 @@ RUN curl -s -L -J https://github.com/DrTimothyAldenDavis/GraphBLAS/archive/refs/ && sudo make install \ && /bin/rm -Rf GraphBLAS-10.2.0 -RUN curl -s -L -J https://github.com/GraphBLAS/LAGraph/archive/refs/tags/v1.2.1.tar.gz | \ - tar zxvf - && cd LAGraph-1.2.1 && \ +# Bump LAGRAPH_CACHE_BUST (or pass --build-arg LAGRAPH_CACHE_BUST=$(date +%s)) to +# force a fresh clone when a branch/PR has new commits but LAGRAPH_REF is unchanged. +ARG LAGRAPH_CACHE_BUST=0 +RUN echo "lagraph cache bust ${LAGRAPH_CACHE_BUST}" && \ + git clone --depth 1 --branch ${LAGRAPH_REF} ${LAGRAPH_REPO} LAGraph && \ + cd LAGraph && \ make library \ && sudo make install \ - && /bin/rm -Rf LAGraph-1.2.1 - + && cd .. \ + && /bin/rm -Rf LAGraph + RUN sudo ldconfig # put test stuff into pg home @@ -88,7 +100,7 @@ ENV EDITOR zile # make the extension #RUN .virt/bin/python3 generate.py onesparse/onesparse--0.1.0.sql -RUN sudo make && sudo make install && sudo make clean +RUN sudo make WITH_MATRIX_SUM=${WITH_MATRIX_SUM} && sudo make install && sudo make clean # start the database RUN initdb -D "$PGDATA" \ diff --git a/Makefile b/Makefile index 5475736..de5f40a 100644 --- a/Makefile +++ b/Makefile @@ -9,6 +9,15 @@ OBJS = $(patsubst %.c,%.o,$(shell find src -name '*.c')) SHLIB_LINK = -lc -lgraphblas -llagraph -llagraphx -lpq PG_CPPFLAGS = -Wfatal-errors -std=c11 -I/usr/local/include/suitesparse/ +# Experimental: build the matrix_sum aggregate, which requires the +# LAGraph_Matrix_Sum primitive from the LAGraph "matrix-sum" branch +# (https://github.com/GraphBLAS/LAGraph/pull/407). Off by default so stock +# LAGraph builds still compile. Enable with: make WITH_MATRIX_SUM=1 +WITH_MATRIX_SUM ?= 0 +ifeq ($(WITH_MATRIX_SUM),1) +PG_CPPFLAGS += -DONESPARSE_HAVE_LAGRAPH_MATRIX_SUM +endif + TESTS = $(wildcard sql/*.sql) REGRESS = $(patsubst sql/%.sql,%,$(TESTS)) # REGRESS_OPTS = --load-language=plpgsql diff --git a/bench/matrix_sum.sql b/bench/matrix_sum.sql new file mode 100644 index 0000000..954c890 --- /dev/null +++ b/bench/matrix_sum.sql @@ -0,0 +1,63 @@ +-- Experimental matrix_sum aggregate. +-- +-- Requires OneSparse built with WITH_MATRIX_SUM=1 against the LAGraph +-- "matrix-sum" branch (https://github.com/GraphBLAS/LAGraph/pull/407), which +-- provides LAGraph_Matrix_Sum. Kept out of the generated extension SQL so stock +-- builds and CREATE EXTENSION are unaffected; load it manually after +-- CREATE EXTENSION onesparse: +-- +-- \i bench/matrix_sum.sql +-- +-- matrix_sum collects the per-row matrix handles during the scan and performs a +-- single LAGraph_Matrix_Sum (extract-tuples + one GrB_Matrix_build) at finalize, +-- combining duplicate (i,j) entries with a dup operator that defaults to PLUS. +-- An optional binaryop argument overrides the dup operator. + +CREATE FUNCTION matrix_sum_matrix(state internal, a matrix) +RETURNS internal +AS '$libdir/onesparse', 'matrix_sum_matrix' +LANGUAGE C; + +CREATE FUNCTION matrix_sum_matrix(state internal, a matrix, op binaryop) +RETURNS internal +AS '$libdir/onesparse', 'matrix_sum_matrix' +LANGUAGE C; + +CREATE FUNCTION matrix_sum_final(state internal) +RETURNS matrix +AS '$libdir/onesparse', 'matrix_sum_final' +LANGUAGE C STRICT; + +CREATE AGGREGATE matrix_sum (a matrix) + (SFUNC=matrix_sum_matrix, STYPE=internal, FINALFUNC=matrix_sum_final); + +CREATE AGGREGATE matrix_sum (a matrix, op binaryop) + (SFUNC=matrix_sum_matrix, STYPE=internal, FINALFUNC=matrix_sum_final); + +-- matrix_binary_sum: same interface and result as matrix_sum, but uses +-- LAGraph_Matrix_Binary_Sum (a different summing technique inside LAGraph) for +-- a different parallel/time/space tradeoff. +CREATE FUNCTION matrix_binary_sum_matrix(state internal, a matrix) +RETURNS internal +AS '$libdir/onesparse', 'matrix_binary_sum_matrix' +LANGUAGE C; + +CREATE FUNCTION matrix_binary_sum_matrix(state internal, a matrix, op binaryop) +RETURNS internal +AS '$libdir/onesparse', 'matrix_binary_sum_matrix' +LANGUAGE C; + +CREATE FUNCTION matrix_binary_sum_final(state internal) +RETURNS matrix +AS '$libdir/onesparse', 'matrix_binary_sum_final' +LANGUAGE C STRICT; + +CREATE AGGREGATE matrix_binary_sum (a matrix) + (SFUNC=matrix_binary_sum_matrix, STYPE=internal, FINALFUNC=matrix_binary_sum_final); + +CREATE AGGREGATE matrix_binary_sum (a matrix, op binaryop) + (SFUNC=matrix_binary_sum_matrix, STYPE=internal, FINALFUNC=matrix_binary_sum_final); + +-- matrix_agg(a matrix) and matrix_agg(a matrix, op binaryop) now ship in the +-- stock extension (binary-counter merge, default dup PLUS), so no bench-only +-- definition is needed here. diff --git a/bench/matrix_sum_bench.sql b/bench/matrix_sum_bench.sql new file mode 100644 index 0000000..904dccf --- /dev/null +++ b/bench/matrix_sum_bench.sql @@ -0,0 +1,49 @@ +-- Benchmark + correctness: matrix_sum (LAGraph single build) vs matrix_binary_sum +-- (LAGraph binary reduction) vs matrix_agg (binary-counter merge). Run manually; +-- not auto-discovered by the regression suite. Requires bench/matrix_sum.sql +-- loaded first (matrix_agg itself ships in the stock extension): +-- +-- CREATE EXTENSION onesparse; +-- \i bench/matrix_sum.sql +-- \i bench/matrix_sum_bench.sql +-- +-- Notes on the inputs: +-- * All engines must see the SAME matrices and require the inputs to share +-- dimensions (a GrB_eWiseAdd / build requirement). We resize the bounded +-- random_matrix output to unbounded with resize(m, -1, -1) so the whole set +-- is uniform. +-- * Use an integer type so PLUS is exactly associative and the results are +-- bit-identical (FP reassociation under build/reduction can differ). +-- * Compare with the matrix = operator, NOT print(): print() cannot render an +-- unbounded matrix (its logical dimension is 2^60). + +\timing on + +-- Correctness on a modest set: the build-based sum and the pairwise PLUS +-- aggregate must agree exactly. +DROP TABLE IF EXISTS bench_small; +CREATE TABLE bench_small AS + SELECT g AS id, resize(random_matrix('int64', 200, 200, 0.05, g), -1, -1) AS a + FROM generate_series(1, 50) g; + +-- All three summing engines must agree exactly (matrix_agg now defaults to PLUS). +SELECT (matrix_agg(a) = matrix_sum(a)) AS agg_eq_sum, + (matrix_sum(a) = matrix_binary_sum(a)) AS sum_eq_binary_sum +FROM bench_small; + +-- Larger fixture for timing. Tune the count / dimensions / density (overlap) +-- to explore where the single internally-parallel build wins. +DROP TABLE IF EXISTS bench_big; +CREATE TABLE bench_big AS + SELECT g AS id, resize(random_matrix('int64', 1000, 1000, 0.005, g), -1, -1) AS a + FROM generate_series(1, 1000) g; + +-- Warm up, then compare wall-clock (\timing). nvals() forces full evaluation +-- while keeping the printed output tiny. +SELECT nvals(matrix_agg(a)) AS agg_nvals FROM bench_big; +SELECT nvals(matrix_sum(a)) AS sum_nvals FROM bench_big; +SELECT nvals(matrix_binary_sum(a)) AS binary_sum_nvals FROM bench_big; + +-- GraphBLAS thread count affects the internally-parallel build. There is no +-- SQL GUC for it; set OMP_NUM_THREADS in the server's environment before +-- starting PostgreSQL and re-run, sweeping e.g. 1, 2, 4, 8 to find the knee. diff --git a/bench/matrix_sum_grb_bench.sql b/bench/matrix_sum_grb_bench.sql new file mode 100644 index 0000000..94ff930 --- /dev/null +++ b/bench/matrix_sum_grb_bench.sql @@ -0,0 +1,66 @@ +-- Low-overlap benchmark: matrix_sum (LAGraph single build) vs matrix_agg +-- (binary-counter merge) over a corpus of serialized GraphBLAS matrices (.grb +-- files produced by GxB_Matrix_serialize), such as ~/dev/NSC/data/grb. +-- +-- Requires OneSparse built with WITH_MATRIX_SUM=1 against the LAGraph +-- "matrix-sum" branch, and bench/matrix_sum.sql loaded first. The .grb corpus +-- must be readable by the server; mount/point it at /data (so /data/manifest.csv +-- and /data//.grb exist), then: +-- +-- CREATE EXTENSION onesparse; +-- \i bench/matrix_sum.sql +-- \i bench/matrix_sum_grb_bench.sql +-- +-- Why resize(..., -1, -1): all engines require the inputs to share dimensions +-- (a GrB_eWiseAdd / build requirement). Resizing every matrix to unbounded +-- (GxB_INDEX_MAX) makes the whole corpus uniform regardless of its stored dims. +-- The corpus is uint32, so the PLUS dup operator is plus_uint32; every engine +-- defaults to PLUS. + +\timing on + +-- The manifest lists every matrix: global_index, inner_tar, n, rel_path, bytes. +DROP TABLE IF EXISTS manifest; +CREATE TABLE manifest(global_index int, inner_tar text, n int, rel_path text, bytes bigint); +COPY manifest FROM '/data/manifest.csv' CSV HEADER; + +-- Load the first :k matrices into a table, resized to unbounded. Set :k on the +-- psql command line, e.g. psql -v k=512 -f bench/matrix_sum_grb_bench.sql +\if :{?k} +\else + \set k 256 +\endif +\echo loading :k matrices + +DROP TABLE IF EXISTS bench; +CREATE TABLE bench AS + SELECT resize(deserialize_file('/data/' || rel_path), -1, -1) AS a + FROM manifest WHERE global_index < :k; + +-- Overlap check: with a low/no-overlap corpus, the union nvals approaches the +-- sum of the per-matrix nvals (so the result is large), while matrix_sum +-- sorts/builds once. +SELECT sum(nvals(a)) AS total_input_nvals, nvals(matrix_sum(a)) AS union_nvals FROM bench; + +-- matrix_sum: one internally-parallel LAGraph build. +-- PEAK MEMORY: LAGraph_Matrix_Sum materializes every input tuple at once +-- (peak ~= sum of all input nvals) before the dedup build. On a zero-overlap +-- corpus this grows linearly with :k; measured on a 39 GB box, k=2048 +-- (268M tuples) succeeded in ~44s but k=4096 (536M tuples) OOM-crashed the +-- backend. Watch RSS and cap :k accordingly, or use the memory-bounded +-- per-worker-build hybrid (see the companion design plan). +\echo -- matrix_sum (LAGraph build) -- +SELECT nvals(matrix_sum(a)) AS sum_nvals FROM bench; + +-- matrix_binary_sum: same result, different LAGraph summing technique +-- (binary reduction tree of eWiseAdds rather than a single concatenate+build). +-- Different parallel/time/space tradeoff; compare its wall time against +-- matrix_sum's above. +\echo -- matrix_binary_sum (LAGraph binary reduction) -- +SELECT nvals(matrix_binary_sum(a)) AS binary_sum_nvals FROM bench; + +-- matrix_agg: serial binary-counter merge (O(N log N), no LAGraph dependency). +-- Single-threaded so slower than the parallel LAGraph engines above, but frees +-- each input as it is absorbed, so it is memory-frugal and reaches high :k. +\echo -- matrix_agg (binary-counter merge) -- +SELECT nvals(matrix_agg(a, 'plus_uint32'::binaryop)) AS agg_nvals FROM bench; diff --git a/bench/run_three_way.sh b/bench/run_three_way.sh new file mode 100755 index 0000000..b3893d0 --- /dev/null +++ b/bench/run_three_way.sh @@ -0,0 +1,43 @@ +#!/bin/bash +# Three-way scaling benchmark over the graphchallenge .grb corpus: +# matrix_agg (serial binary-counter merge) vs matrix_sum (single concat+build) +# vs matrix_binary_sum (binary reduction tree). matrix_agg is the speedup +# baseline; AGG_MAX caps how high it is run (it is single-threaded, so slower +# than the parallel LAGraph engines). Assumes a table allm(idx, a) is loaded and the +# matrix_sum.sql aggregates are registered. Each query runs ONCE; both its +# result (nvals) and elapsed time are captured from that single run. +set -u +DB=onesparse-bench-db +# matrix_agg is now an O(N log N) binary-counter merge, so it can run across the +# whole sweep; AGG_MAX caps it only if you want to bound wall time. Override from +# the environment, e.g. AGG_MAX=4096 ./run_three_way.sh +AGG_MAX="${AGG_MAX:-4096}" +KS="${KS:-64 128 256 512 1024 2048 4096}" + +# run one timed query; print "|", or "CRASH|CRASH" if the backend dies +run() { + local sql="$1" out nv ms + out=$(docker exec -i "$DB" psql -U postgres -t -A -c "\timing on" -c "$sql" 2>&1) + if echo "$out" | grep -q "server closed the connection\|terminating connection\|could not connect\|out of memory"; then + until docker exec "$DB" psql -o /dev/null -t -q -U postgres -c 'select 1' 2>/dev/null; do sleep 1; done + echo "CRASH|CRASH"; return + fi + nv=$(echo "$out" | grep -vE "Time:|Timing is on\.|^$" | head -1) + ms=$(echo "$out" | grep -oE "Time: [0-9.]+ ms" | grep -oE "[0-9.]+" | head -1) + echo "${nv:-?}|${ms:-?}" +} + +echo "K|agg_ms|sum_ms|binsum_ms|agg_nvals|sum_nvals|binsum_nvals" +for K in $KS; do + W="FROM allm WHERE idx < $K" + + IFS='|' read sum_n sum_t <<< "$(run "SELECT nvals(matrix_sum(a)) $W")" + IFS='|' read bin_n bin_t <<< "$(run "SELECT nvals(matrix_binary_sum(a)) $W")" + if [ "$K" -le "$AGG_MAX" ]; then + IFS='|' read agg_n agg_t <<< "$(run "SELECT nvals(matrix_agg(a,'plus_uint32'::binaryop)) $W")" + else + agg_n="—"; agg_t="—" + fi + + echo "$K|$agg_t|$sum_t|$bin_t|$agg_n|$sum_n|$bin_n" +done diff --git a/expected/test_guc.out b/expected/test_guc.out new file mode 100644 index 0000000..dd1bacb --- /dev/null +++ b/expected/test_guc.out @@ -0,0 +1,72 @@ +set client_min_messages = 'WARNING'; +create extension if not exists onesparse; +set search_path = 'onesparse'; +-- verify defaults +show onesparse.nthreads; + onesparse.nthreads +-------------------- + 0 +(1 row) + +show onesparse.chunk; + onesparse.chunk +----------------- + 0 +(1 row) + +show onesparse.print_1based; + onesparse.print_1based +------------------------ + off +(1 row) + +show onesparse.hyper_switch; + onesparse.hyper_switch +------------------------ + 0.0625 +(1 row) + +show onesparse.jit_use_cmake; + onesparse.jit_use_cmake +------------------------- + off +(1 row) + +-- set and verify session-level GUCs +set onesparse.nthreads = 4; +show onesparse.nthreads; + onesparse.nthreads +-------------------- + 4 +(1 row) + +set onesparse.print_1based = on; +show onesparse.print_1based; + onesparse.print_1based +------------------------ + on +(1 row) + +set onesparse.hyper_switch = 0.5; +show onesparse.hyper_switch; + onesparse.hyper_switch +------------------------ + 0.5 +(1 row) + +-- reset to defaults +reset onesparse.nthreads; +reset onesparse.print_1based; +reset onesparse.hyper_switch; +show onesparse.nthreads; + onesparse.nthreads +-------------------- + 0 +(1 row) + +show onesparse.hyper_switch; + onesparse.hyper_switch +------------------------ + 0.0625 +(1 row) + diff --git a/expected/test_matrix_header.out b/expected/test_matrix_header.out index e75d67a..be3cfdc 100644 --- a/expected/test_matrix_header.out +++ b/expected/test_matrix_header.out @@ -5539,6 +5539,61 @@ HINT: No function matches the given name and argument types. You might need to drop table mat_agg_test; drop table mat_agg_float; drop table mat_agg_bool; +-- ## Combining matrices with `matrix_agg(a matrix)` +-- +-- The matrix-combining form of `matrix_agg` sums one matrix per row into a +-- single result using a binary-counter merge. The default duplicate combiner is +-- PLUS, so overlapping entries are added (matching `matrix_sum`). +create temporary table mat_combine (a matrix); +insert into mat_combine values + ('int64(3:3)[0:0:1 1:1:2]'::matrix), + ('int64(3:3)[0:0:10 2:2:3]'::matrix), + ('int64(3:3)[0:0:100 1:1:20]'::matrix); +-- Default (PLUS): (0,0)=111, (1,1)=22, (2,2)=3 +select print(matrix_agg(a)) as combined from mat_combine; +┌──────────────────────┐ +│ combined │ +├──────────────────────┤ +│ 0 1 2 ↵│ +│ ──────────── ↵│ +│ 0│ 111 ↵│ +│ 1│ 22 ↵│ +│ 2│ 3 ↵│ +│ │ +└──────────────────────┘ +(1 row) + +-- The default equals an explicit PLUS, and matches the expected literal +select matrix_agg(a) = matrix_agg(a, 'plus_int64'::binaryop) as default_is_plus, + matrix_agg(a) = 'int64(3:3)[0:0:111 1:1:22 2:2:3]'::matrix as matches_expected +from mat_combine; +┌─────────────────┬──────────────────┐ +│ default_is_plus │ matches_expected │ +├─────────────────┼──────────────────┤ +│ t │ t │ +└─────────────────┴──────────────────┘ +(1 row) + +-- A non-default combiner (MAX): (0,0)=100, (1,1)=20, (2,2)=3 +select matrix_agg(a, 'max_int64'::binaryop) = 'int64(3:3)[0:0:100 1:1:20 2:2:3]'::matrix as max_ok +from mat_combine; +┌────────┐ +│ max_ok │ +├────────┤ +│ t │ +└────────┘ +(1 row) + +-- Empty input aggregates to NULL +select matrix_agg(a) is null as empty_is_null from mat_combine where false; +┌───────────────┐ +│ empty_is_null │ +├───────────────┤ +│ t │ +└───────────────┘ +(1 row) + +drop table mat_combine; -- ## Descriptor Basics -- -- Test basic descriptor construction and properties. diff --git a/onesparse/onesparse--1.0.1.sql b/onesparse/onesparse--1.0.1.sql index f058e0b..79f47dc 100644 --- a/onesparse/onesparse--1.0.1.sql +++ b/onesparse/onesparse--1.0.1.sql @@ -2139,21 +2139,44 @@ RETURNS matrix AS '$libdir/onesparse', 'matrix_cast' LANGUAGE C STABLE; -CREATE FUNCTION matrix_agg_matrix(state matrix, a matrix) +-- matrix_agg_final is the finalfunc for the per-type element builder +-- matrix_agg(i, j, v) (see matrix_op.sql); leave it as-is. +CREATE FUNCTION matrix_agg_final(matrix) RETURNS matrix +AS '$libdir/onesparse', 'matrix_agg_final' +LANGUAGE C STRICT; + +-- The matrix-combining matrix_agg(a matrix) uses a binary-counter merge with an +-- internal carry-stack state (src/matrix/matrix_agg_matrix.c). The optional +-- binaryop selects the duplicate combiner; the default is PLUS, matching +-- matrix_sum and matrix_binary_sum. +CREATE FUNCTION matrix_agg_matrix(state internal, a matrix) +RETURNS internal AS '$libdir/onesparse', 'matrix_agg_matrix' -LANGUAGE C STABLE; +LANGUAGE C; -CREATE FUNCTION matrix_agg_final(matrix) +CREATE FUNCTION matrix_agg_matrix(state internal, a matrix, op binaryop) +RETURNS internal +AS '$libdir/onesparse', 'matrix_agg_matrix' +LANGUAGE C; + +CREATE FUNCTION matrix_agg_combine_final(state internal) RETURNS matrix -AS '$libdir/onesparse', 'matrix_agg_final' +AS '$libdir/onesparse', 'matrix_agg_combine_final' LANGUAGE C STRICT; CREATE AGGREGATE matrix_agg (a matrix) ( SFUNC=matrix_agg_matrix, - STYPE=matrix, - FINALFUNC=matrix_agg_final + STYPE=internal, + FINALFUNC=matrix_agg_combine_final + ); + +CREATE AGGREGATE matrix_agg (a matrix, op binaryop) + ( + SFUNC=matrix_agg_matrix, + STYPE=internal, + FINALFUNC=matrix_agg_combine_final ); CREATE FUNCTION set_element(a matrix, i bigint, j bigint, s scalar) diff --git a/sql/test_guc.sql b/sql/test_guc.sql new file mode 100644 index 0000000..a9f5679 --- /dev/null +++ b/sql/test_guc.sql @@ -0,0 +1,27 @@ +set client_min_messages = 'WARNING'; +create extension if not exists onesparse; +set search_path = 'onesparse'; + +-- verify defaults +show onesparse.nthreads; +show onesparse.chunk; +show onesparse.print_1based; +show onesparse.hyper_switch; +show onesparse.jit_use_cmake; + +-- set and verify session-level GUCs +set onesparse.nthreads = 4; +show onesparse.nthreads; + +set onesparse.print_1based = on; +show onesparse.print_1based; + +set onesparse.hyper_switch = 0.5; +show onesparse.hyper_switch; + +-- reset to defaults +reset onesparse.nthreads; +reset onesparse.print_1based; +reset onesparse.hyper_switch; +show onesparse.nthreads; +show onesparse.hyper_switch; diff --git a/sql/test_matrix_header.sql b/sql/test_matrix_header.sql index 2a6cf6e..eb9aa4d 100644 --- a/sql/test_matrix_header.sql +++ b/sql/test_matrix_header.sql @@ -769,6 +769,35 @@ drop table mat_agg_test; drop table mat_agg_float; drop table mat_agg_bool; +-- ## Combining matrices with `matrix_agg(a matrix)` +-- +-- The matrix-combining form of `matrix_agg` sums one matrix per row into a +-- single result using a binary-counter merge. The default duplicate combiner is +-- PLUS, so overlapping entries are added (matching `matrix_sum`). + +create temporary table mat_combine (a matrix); +insert into mat_combine values + ('int64(3:3)[0:0:1 1:1:2]'::matrix), + ('int64(3:3)[0:0:10 2:2:3]'::matrix), + ('int64(3:3)[0:0:100 1:1:20]'::matrix); + +-- Default (PLUS): (0,0)=111, (1,1)=22, (2,2)=3 +select print(matrix_agg(a)) as combined from mat_combine; + +-- The default equals an explicit PLUS, and matches the expected literal +select matrix_agg(a) = matrix_agg(a, 'plus_int64'::binaryop) as default_is_plus, + matrix_agg(a) = 'int64(3:3)[0:0:111 1:1:22 2:2:3]'::matrix as matches_expected +from mat_combine; + +-- A non-default combiner (MAX): (0,0)=100, (1,1)=20, (2,2)=3 +select matrix_agg(a, 'max_int64'::binaryop) = 'int64(3:3)[0:0:100 1:1:20 2:2:3]'::matrix as max_ok +from mat_combine; + +-- Empty input aggregates to NULL +select matrix_agg(a) is null as empty_is_null from mat_combine where false; + +drop table mat_combine; + -- ## Descriptor Basics -- -- Test basic descriptor construction and properties. diff --git a/src/guc.c b/src/guc.c index 47a47d2..28f47e2 100644 --- a/src/guc.c +++ b/src/guc.c @@ -6,6 +6,20 @@ static bool os_burble = false; static char *os_jit_control = NULL; static GxB_JIT_Control os_jit_control_mode = GxB_JIT_OFF; +static int os_nthreads = 0; +static double os_chunk = 0.0; +static bool os_print_1based = false; +static double os_hyper_switch = 0.0625; + +static char *os_jit_cache_path = NULL; +static char *os_jit_c_compiler_name = NULL; +static char *os_jit_c_compiler_flags = NULL; +static char *os_jit_c_linker_flags = NULL; +static char *os_jit_c_libraries = NULL; +static char *os_jit_c_preface = NULL; +static char *os_jit_error_log = NULL; +static bool os_jit_use_cmake = false; + static void burble_assign_hook(bool newvalue, void *extra) { @@ -108,6 +122,99 @@ check_os_jit_control(char **newval, void **extra, GucSource source) return true; } +static void +nthreads_assign_hook(int newvalue, void *extra) +{ + if (newvalue == 0) return; /* 0 = let GraphBLAS/OpenMP decide */ + OK_CHECK(GrB_set(GrB_GLOBAL, (int32_t) newvalue, GxB_GLOBAL_NTHREADS), + "Cannot set nthreads"); +} + +static void +chunk_assign_hook(double newvalue, void *extra) +{ + if (newvalue == 0.0) return; /* 0 = let GraphBLAS decide */ + OK_CHECK(GxB_Global_Option_set_FP64(GxB_CHUNK, newvalue), + "Cannot set chunk"); +} + +static void +print_1based_assign_hook(bool newvalue, void *extra) +{ + OK_CHECK(GrB_set(GrB_GLOBAL, (int32_t) newvalue, GxB_PRINT_1BASED), + "Cannot set print_1based"); +} + +static void +hyper_switch_assign_hook(double newvalue, void *extra) +{ + OK_CHECK(GxB_Global_Option_set_FP64(GxB_HYPER_SWITCH, newvalue), + "Cannot set hyper_switch"); +} + +static void +jit_cache_path_assign_hook(const char *newval, void *extra) +{ + if (newval == NULL) return; + OK_CHECK(GxB_Global_Option_set_CHAR(GxB_JIT_CACHE_PATH, newval), + "Cannot set JIT cache path"); +} + +static void +jit_c_compiler_name_assign_hook(const char *newval, void *extra) +{ + if (newval == NULL) return; + OK_CHECK(GxB_Global_Option_set_CHAR(GxB_JIT_C_COMPILER_NAME, newval), + "Cannot set JIT C compiler name"); +} + +static void +jit_c_compiler_flags_assign_hook(const char *newval, void *extra) +{ + if (newval == NULL) return; + OK_CHECK(GxB_Global_Option_set_CHAR(GxB_JIT_C_COMPILER_FLAGS, newval), + "Cannot set JIT C compiler flags"); +} + +static void +jit_c_linker_flags_assign_hook(const char *newval, void *extra) +{ + if (newval == NULL) return; + OK_CHECK(GxB_Global_Option_set_CHAR(GxB_JIT_C_LINKER_FLAGS, newval), + "Cannot set JIT C linker flags"); +} + +static void +jit_c_libraries_assign_hook(const char *newval, void *extra) +{ + if (newval == NULL) return; + OK_CHECK(GxB_Global_Option_set_CHAR(GxB_JIT_C_LIBRARIES, newval), + "Cannot set JIT C libraries"); +} + +static void +jit_c_preface_assign_hook(const char *newval, void *extra) +{ + if (newval == NULL) return; + OK_CHECK(GxB_Global_Option_set_CHAR(GxB_JIT_C_PREFACE, newval), + "Cannot set JIT C preface"); +} + +static void +jit_error_log_assign_hook(const char *newval, void *extra) +{ + if (newval == NULL) return; + OK_CHECK(GxB_Global_Option_set_CHAR(GxB_JIT_ERROR_LOG, newval), + "Cannot set JIT error log"); +} + +static void +jit_use_cmake_assign_hook(bool newvalue, void *extra) +{ + OK_CHECK(GrB_set(GrB_GLOBAL, (int32_t) newvalue, GxB_JIT_USE_CMAKE), + "Cannot set JIT use cmake"); +} + void initialize_gucs() { @@ -136,6 +243,155 @@ initialize_gucs() 0, check_os_jit_control, NULL, - NULL - ); + NULL); + + DefineCustomIntVariable( + "onesparse.nthreads", + "Maximum number of threads for SuiteSparse:GraphBLAS operations.", + "0 means use all available threads (OpenMP default).", + &os_nthreads, + 0, + 0, + 65536, + PGC_USERSET, + 0, + NULL, + nthreads_assign_hook, + NULL); + + DefineCustomRealVariable( + "onesparse.chunk", + "Chunk size for small problems in SuiteSparse:GraphBLAS.", + "0 means automatic (GraphBLAS default). Typical values: 65536, 262144.", + &os_chunk, + 0.0, + 0.0, + 1e15, + PGC_USERSET, + 0, + NULL, + chunk_assign_hook, + NULL); + + DefineCustomBoolVariable( + "onesparse.print_1based", + "Print matrix/vector indices as 1-based instead of 0-based.", + NULL, + &os_print_1based, + false, + PGC_USERSET, + 0, + NULL, + print_1based_assign_hook, + NULL); + + DefineCustomRealVariable( + "onesparse.hyper_switch", + "Threshold to switch between hypersparse and sparse format.", + "Use GxB_ALWAYS_HYPER (-1) or GxB_NEVER_HYPER (0). Default ~0.0625.", + &os_hyper_switch, + 0.0625, + -1.0, + 1.0, + PGC_USERSET, + 0, + NULL, + hyper_switch_assign_hook, + NULL); + + DefineCustomStringVariable( + "onesparse.jit_cache_path", + "Cache directory for SuiteSparse JIT compiled kernels.", + NULL, + &os_jit_cache_path, + NULL, + PGC_POSTMASTER, + 0, + NULL, + jit_cache_path_assign_hook, + NULL); + + DefineCustomStringVariable( + "onesparse.jit_c_compiler_name", + "C compiler for SuiteSparse JIT (e.g. \"gcc\").", + NULL, + &os_jit_c_compiler_name, + NULL, + PGC_POSTMASTER, + 0, + NULL, + jit_c_compiler_name_assign_hook, + NULL); + + DefineCustomStringVariable( + "onesparse.jit_c_compiler_flags", + "Compiler flags for SuiteSparse JIT.", + NULL, + &os_jit_c_compiler_flags, + NULL, + PGC_POSTMASTER, + 0, + NULL, + jit_c_compiler_flags_assign_hook, + NULL); + + DefineCustomStringVariable( + "onesparse.jit_c_linker_flags", + "Linker flags for SuiteSparse JIT.", + NULL, + &os_jit_c_linker_flags, + NULL, + PGC_POSTMASTER, + 0, + NULL, + jit_c_linker_flags_assign_hook, + NULL); + + DefineCustomStringVariable( + "onesparse.jit_c_libraries", + "Libraries for SuiteSparse JIT.", + NULL, + &os_jit_c_libraries, + NULL, + PGC_POSTMASTER, + 0, + NULL, + jit_c_libraries_assign_hook, + NULL); + + DefineCustomStringVariable( + "onesparse.jit_c_preface", + "C code preface inserted into every SuiteSparse JIT kernel.", + NULL, + &os_jit_c_preface, + NULL, + PGC_POSTMASTER, + 0, + NULL, + jit_c_preface_assign_hook, + NULL); + + DefineCustomStringVariable( + "onesparse.jit_error_log", + "File path for SuiteSparse JIT compilation error log.", + NULL, + &os_jit_error_log, + NULL, + PGC_POSTMASTER, + 0, + NULL, + jit_error_log_assign_hook, + NULL); + + DefineCustomBoolVariable( + "onesparse.jit_use_cmake", + "Use cmake instead of direct compile for SuiteSparse JIT.", + NULL, + &os_jit_use_cmake, + false, + PGC_POSTMASTER, + 0, + NULL, + jit_use_cmake_assign_hook, + NULL); } diff --git a/src/matrix/matrix_agg_combine_final.c b/src/matrix/matrix_agg_combine_final.c new file mode 100644 index 0000000..e6840cd --- /dev/null +++ b/src/matrix/matrix_agg_combine_final.c @@ -0,0 +1,63 @@ +#include "../onesparse.h" + +/* + * Finalfunc for the matrix-combining matrix_agg(a matrix) binary-counter merge + * (sfunc: matrix_agg_matrix). Folds the occupied carry levels into a single + * result. Each taken level is stolen out of the state (slot set to NULL) so the + * aggregate context reset callback won't double-free it; the returned C is owned + * by the new os_Matrix. + * + * The per-type element builder matrix_agg(i, j, v) uses a different finalfunc + * (matrix_agg_final) and is unaffected by this. + */ + +/* mirror of the MatrixAggState in matrix_agg_matrix.c */ +#define MATRIX_AGG_MAX_LEVELS 64 + +typedef struct MatrixAggState +{ + GrB_Matrix levels[MATRIX_AGG_MAX_LEVELS]; + GrB_Type type; + GrB_BinaryOp dup; +} MatrixAggState; + +PG_FUNCTION_INFO_V1(matrix_agg_combine_final); +Datum +matrix_agg_combine_final(PG_FUNCTION_ARGS) +{ + MatrixAggState *state; + GrB_Matrix C = NULL; + int r; + + if (PG_ARGISNULL(0)) + PG_RETURN_NULL(); + + state = (MatrixAggState *) PG_GETARG_POINTER(0); + + for (r = 0; r < MATRIX_AGG_MAX_LEVELS; r++) + { + if (state->levels[r] == NULL) + continue; + + if (C == NULL) + { + C = state->levels[r]; + state->levels[r] = NULL; + } + else + { + OS_CHECK(GrB_eWiseAdd(C, NULL, NULL, state->dup, + C, state->levels[r], NULL), + C, + "Error matrix eadd in matrix_agg_combine_final."); + OS_CHECK(GrB_Matrix_free(&state->levels[r]), + state->levels[r], + "Cannot GrB_Free matrix_agg carry matrix"); + } + } + + if (C == NULL) + PG_RETURN_NULL(); + + OS_RETURN_MATRIX(new_matrix(NULL, 0, 0, CurrentMemoryContext, C)); +} diff --git a/src/matrix/matrix_agg_matrix.c b/src/matrix/matrix_agg_matrix.c index c0951f6..2a15024 100644 --- a/src/matrix/matrix_agg_matrix.c +++ b/src/matrix/matrix_agg_matrix.c @@ -1,53 +1,142 @@ #include "../onesparse.h" +/* + * matrix_agg(a matrix): an aggregate that combines many matrices (one per row) + * into a single result using a binary-counter merge. + * + * This is the portable sibling of matrix_sum / matrix_binary_sum: same result, + * but implemented purely with GrB_eWiseAdd (no LAGraph dependency, so it ships + * in the stock extension and is not gated). It is single-threaded but streams: + * each input is freed the moment it is absorbed, so it never holds all N inputs + * at once. + * + * The merge is a balanced binary reduction maintained as a binary counter: + * levels[r], when non-NULL, holds the sum of exactly 2^r already-seen matrices + * (like binary addition with carry). Each new matrix enters at rank 0 and is + * carried up, summing-and-freeing occupied levels, until it lands in an empty + * slot. At finalize the occupied levels are folded into a single result. This is + * O(N log N) total eWiseAdd work, versus the old serial accumulator which was + * ~O(N^2) on low-overlap data (it re-touched the whole growing accumulator on + * every row). + * + * The dup operator defaults to PLUS (so matrix_agg(m) is the element-wise sum, + * matching matrix_sum and matrix_binary_sum); an optional binaryop argument + * overrides it. dup must be commutative and associative for the result to be + * well defined (the binary tree then yields the same result as a serial fold). + * All inputs must share dimensions (a GrB_eWiseAdd requirement). + * + * NB: this is the matrix-combining matrix_agg(a matrix). The per-type element + * builder matrix_agg(i, j, v) is a separate aggregate (src/matrix/matrix_ops.h, + * finalfunc matrix_agg_final) and is unaffected. + */ + +#define MATRIX_AGG_MAX_LEVELS 64 + +typedef struct MatrixAggState +{ + GrB_Matrix levels[MATRIX_AGG_MAX_LEVELS]; /* carry slots: levels[r] = sum of 2^r matrices, or NULL */ + GrB_Type type; /* element type, fixed by the first row */ + GrB_BinaryOp dup; /* duplicate combiner for the sum */ +} MatrixAggState; + +/* + * Free any carry matrices still held when the aggregate's memory context is + * reset/deleted. GraphBLAS objects are malloc'd (not in a PG context), so they + * must be freed explicitly to avoid leaks (including on error/abort). Slots the + * finalfunc has already taken ownership of are set to NULL and skipped here. + */ +static void +context_callback_matrix_agg_free(void *ptr) +{ + MatrixAggState *state = (MatrixAggState *) ptr; + int r; + + for (r = 0; r < MATRIX_AGG_MAX_LEVELS; r++) + { + if (state->levels[r] != NULL) + OS_CHECK(GrB_Matrix_free(&state->levels[r]), + state->levels[r], + "Cannot GrB_Free matrix_agg carry matrix"); + } +} + PG_FUNCTION_INFO_V1(matrix_agg_matrix); Datum matrix_agg_matrix(PG_FUNCTION_ARGS) { GrB_Type type; - os_Matrix *state, - *A; + os_Matrix *A; + GrB_Matrix carry; GrB_BinaryOp op; + MatrixAggState *state; MemoryContext aggcontext; int nargs; + int r; if (PG_ARGISNULL(1)) { elog(ERROR, "Cannot aggregate a null value to a matrix"); } + if (!AggCheckCallContext(fcinfo, &aggcontext)) + { + elog(ERROR, "aggregate function called in non-aggregate context"); + } + A = OS_GETARG_MATRIX(1); OS_MTYPE(type, A); if (PG_ARGISNULL(0)) { - if (!AggCheckCallContext(fcinfo, &aggcontext)) - { - elog(ERROR, "aggregate function called in non-aggregate context"); - } - state = new_matrix(type, GxB_INDEX_MAX, GxB_INDEX_MAX, aggcontext, NULL); + MemoryContextCallback *ctxcb; + + state = MemoryContextAllocZero(aggcontext, sizeof(MatrixAggState)); + state->type = type; + + nargs = PG_NARGS(); + op = OS_GETARG_BINARYOP_HANDLE_OR_NULL(nargs, 2); + if (op == NULL) + op = plus_binaryop(type); + state->dup = op; + + ctxcb = MemoryContextAlloc(aggcontext, sizeof(MemoryContextCallback)); + ctxcb->func = context_callback_matrix_agg_free; + ctxcb->arg = state; + MemoryContextRegisterResetCallback(aggcontext, ctxcb); } else { - state = OS_GETARG_MATRIX(0); + state = (MatrixAggState *) PG_GETARG_POINTER(0); } - nargs = PG_NARGS(); - op = OS_GETARG_BINARYOP_HANDLE_OR_NULL(nargs, 2); - if (op == NULL) + /* + * The input matrix lives in a per-row memory context that is reset before + * finalize, so we must dup it into a handle we own for the whole aggregate. + */ + OS_CHECK(GrB_Matrix_dup(&carry, A->matrix), + A->matrix, + "Cannot dup matrix in matrix_agg_matrix"); + + /* + * Carry the new matrix up the counter: while a level is occupied, sum it + * into carry (C may alias A in GrB_eWiseAdd) and free the absorbed level, + * until carry lands in an empty slot. + */ + for (r = 0; r < MATRIX_AGG_MAX_LEVELS && state->levels[r] != NULL; r++) { - op = default_binaryop(type); + OS_CHECK(GrB_eWiseAdd(carry, NULL, NULL, state->dup, + carry, state->levels[r], NULL), + carry, + "Error matrix eadd in matrix_agg_matrix."); + OS_CHECK(GrB_Matrix_free(&state->levels[r]), + state->levels[r], + "Cannot GrB_Free matrix_agg carry matrix"); } - OS_CHECK(GrB_eWiseAdd(state->matrix, - NULL, - NULL, - op, - state->matrix, - A->matrix, - NULL), - state->matrix, - "Error matrix eadd in matrix_agg_matrix."); - - OS_RETURN_MATRIX(state); + if (r == MATRIX_AGG_MAX_LEVELS) + elog(ERROR, "matrix_agg carry overflow (too many input matrices)"); + + state->levels[r] = carry; + + PG_RETURN_POINTER(state); } diff --git a/src/matrix/matrix_binary_sum.c b/src/matrix/matrix_binary_sum.c new file mode 100644 index 0000000..2f44b38 --- /dev/null +++ b/src/matrix/matrix_binary_sum.c @@ -0,0 +1,161 @@ +#include "../onesparse.h" + +/* + * matrix_binary_sum: an aggregate that combines many matrices (one per row) + * into a single result using LAGraph_Matrix_Binary_Sum. + * + * This is the sibling of matrix_sum (src/matrix/matrix_sum.c): same interface, + * same result, different summing technique inside LAGraph. Where + * LAGraph_Matrix_Sum concatenates the tuples of all inputs and does a single + * GrB_Matrix_build with a dup operator, LAGraph_Matrix_Binary_Sum combines the + * inputs with a different strategy (a binary reduction tree of GrB_eWiseAdds); + * both yield identical results for a commutative+associative dup operator but + * trade off parallelism, time, and peak memory differently. matrix_agg + * (a serial binary-counter merge of GrB_eWiseAdds, no LAGraph dependency) is the + * portable baseline; this brings the family of summing aggregates to three. + * + * The dup operator defaults to PLUS (so matrix_binary_sum(m) is the element-wise + * sum); an optional binaryop argument overrides it. dup must be commutative and + * associative for the result to be well defined. + * + * Builtin types only: LAGraph_Matrix_Binary_Sum returns GrB_NOT_IMPLEMENTED for + * user-defined types. + * + * Gated behind ONESPARSE_HAVE_LAGRAPH_MATRIX_SUM (same flag as matrix_sum) + * because LAGraph_Matrix_Binary_Sum only exists on the LAGraph "matrix-sum" + * branch (https://github.com/GraphBLAS/LAGraph/pull/407). Build with + * `make WITH_MATRIX_SUM=1` against that branch to enable it. + */ + +#ifdef ONESPARSE_HAVE_LAGRAPH_MATRIX_SUM + +typedef struct MatrixBinarySumState +{ + GrB_Matrix *mats; /* dup'd input handles, grown geometrically */ + int n; /* number of handles collected */ + int cap; /* allocated slots in mats */ + GrB_Type type; /* element type, fixed by the first row */ + GrB_BinaryOp dup; /* duplicate combiner for the final sum */ +} MatrixBinarySumState; + +/* + * Free the collected dup'd input matrices when the aggregate's memory context + * is reset/deleted. GraphBLAS objects are malloc'd (not in a PG context), so + * they must be freed explicitly to avoid leaks (including on error/abort). + */ +static void +context_callback_matrix_binary_sum_free(void *ptr) +{ + MatrixBinarySumState *state = (MatrixBinarySumState *) ptr; + int i; + + for (i = 0; i < state->n; i++) + { + if (state->mats[i] != NULL) + OS_CHECK(GrB_Matrix_free(&state->mats[i]), + state->mats[i], + "Cannot GrB_Free matrix_binary_sum input matrix"); + } +} + +PG_FUNCTION_INFO_V1(matrix_binary_sum_matrix); +Datum +matrix_binary_sum_matrix(PG_FUNCTION_ARGS) +{ + GrB_Type type; + os_Matrix *A; + GrB_Matrix dupmat; + GrB_BinaryOp op; + MatrixBinarySumState *state; + MemoryContext aggcontext; + int nargs; + + if (PG_ARGISNULL(1)) + { + elog(ERROR, "Cannot aggregate a null value to a matrix"); + } + + if (!AggCheckCallContext(fcinfo, &aggcontext)) + { + elog(ERROR, "aggregate function called in non-aggregate context"); + } + + A = OS_GETARG_MATRIX(1); + OS_MTYPE(type, A); + + if (PG_ARGISNULL(0)) + { + MemoryContextCallback *ctxcb; + + state = MemoryContextAllocZero(aggcontext, sizeof(MatrixBinarySumState)); + state->cap = 16; + state->mats = MemoryContextAlloc(aggcontext, + state->cap * sizeof(GrB_Matrix)); + state->n = 0; + state->type = type; + + nargs = PG_NARGS(); + op = OS_GETARG_BINARYOP_HANDLE_OR_NULL(nargs, 2); + if (op == NULL) + op = plus_binaryop(type); + state->dup = op; + + ctxcb = MemoryContextAlloc(aggcontext, sizeof(MemoryContextCallback)); + ctxcb->func = context_callback_matrix_binary_sum_free; + ctxcb->arg = state; + MemoryContextRegisterResetCallback(aggcontext, ctxcb); + } + else + { + state = (MatrixBinarySumState *) PG_GETARG_POINTER(0); + } + + if (state->n == state->cap) + { + state->cap *= 2; + state->mats = repalloc(state->mats, + state->cap * sizeof(GrB_Matrix)); + } + + /* + * The input matrix lives in a per-row memory context that is reset before + * finalize, so we must dup it into a handle we own for the whole aggregate. + */ + OS_CHECK(GrB_Matrix_dup(&dupmat, A->matrix), + A->matrix, + "Cannot dup matrix in matrix_binary_sum_matrix"); + state->mats[state->n++] = dupmat; + + PG_RETURN_POINTER(state); +} + +PG_FUNCTION_INFO_V1(matrix_binary_sum_final); +Datum +matrix_binary_sum_final(PG_FUNCTION_ARGS) +{ + MatrixBinarySumState *state; + GrB_Matrix C; + char msg[LAGRAPH_MSG_LEN]; + + if (PG_ARGISNULL(0)) + PG_RETURN_NULL(); + + state = (MatrixBinarySumState *) PG_GETARG_POINTER(0); + + if (state->n == 0) + PG_RETURN_NULL(); + + LA_CHECK(LAGraph_Matrix_Binary_Sum(&C, + state->mats, + (GrB_Index) state->n, + state->dup, + msg)); + + /* + * The dup'd inputs in state->mats are freed by the aggregate context reset + * callback; C is owned by the returned os_Matrix. + */ + OS_RETURN_MATRIX(new_matrix(NULL, 0, 0, CurrentMemoryContext, C)); +} + +#endif /* ONESPARSE_HAVE_LAGRAPH_MATRIX_SUM */ diff --git a/src/matrix/matrix_sum.c b/src/matrix/matrix_sum.c new file mode 100644 index 0000000..843519e --- /dev/null +++ b/src/matrix/matrix_sum.c @@ -0,0 +1,160 @@ +#include "../onesparse.h" + +/* + * matrix_sum: an aggregate that combines many matrices (one per row) into a + * single result using LAGraph_Matrix_Sum. + * + * Unlike matrix_agg, which combines the rows with a serial binary-counter merge + * of GrB_eWiseAdds (no LAGraph dependency), matrix_sum collects the input matrix + * handles during the scan and performs a + * single LAGraph_Matrix_Sum at finalize. That primitive concatenates the tuples + * of all inputs into one buffer and calls GrB_Matrix_build once with a "dup" + * binary operator to combine duplicate (i,j) entries -- a single internally + * parallel build pass instead of N-1 sparse merges. + * + * The dup operator defaults to PLUS (so matrix_sum(m) is the element-wise sum); + * an optional binaryop argument overrides it. dup must be commutative and + * associative for the result to be well defined, since the tuple order seen by + * GrB_Matrix_build is unspecified. + * + * Builtin types only: LAGraph_Matrix_Sum returns GrB_NOT_IMPLEMENTED for + * user-defined types. + * + * This whole file is gated behind ONESPARSE_HAVE_LAGRAPH_MATRIX_SUM because + * LAGraph_Matrix_Sum only exists on the LAGraph "matrix-sum" branch + * (https://github.com/GraphBLAS/LAGraph/pull/407). Build with + * `make WITH_MATRIX_SUM=1` against that branch to enable it. + */ + +#ifdef ONESPARSE_HAVE_LAGRAPH_MATRIX_SUM + +typedef struct MatrixSumState +{ + GrB_Matrix *mats; /* dup'd input handles, grown geometrically */ + int n; /* number of handles collected */ + int cap; /* allocated slots in mats */ + GrB_Type type; /* element type, fixed by the first row */ + GrB_BinaryOp dup; /* duplicate combiner for the final build */ +} MatrixSumState; + +/* + * Free the collected dup'd input matrices when the aggregate's memory context + * is reset/deleted. GraphBLAS objects are malloc'd (not in a PG context), so + * they must be freed explicitly to avoid leaks (including on error/abort). + */ +static void +context_callback_matrix_sum_free(void *ptr) +{ + MatrixSumState *state = (MatrixSumState *) ptr; + int i; + + for (i = 0; i < state->n; i++) + { + if (state->mats[i] != NULL) + OS_CHECK(GrB_Matrix_free(&state->mats[i]), + state->mats[i], + "Cannot GrB_Free matrix_sum input matrix"); + } +} + +PG_FUNCTION_INFO_V1(matrix_sum_matrix); +Datum +matrix_sum_matrix(PG_FUNCTION_ARGS) +{ + GrB_Type type; + os_Matrix *A; + GrB_Matrix dupmat; + GrB_BinaryOp op; + MatrixSumState *state; + MemoryContext aggcontext; + int nargs; + + if (PG_ARGISNULL(1)) + { + elog(ERROR, "Cannot aggregate a null value to a matrix"); + } + + if (!AggCheckCallContext(fcinfo, &aggcontext)) + { + elog(ERROR, "aggregate function called in non-aggregate context"); + } + + A = OS_GETARG_MATRIX(1); + OS_MTYPE(type, A); + + if (PG_ARGISNULL(0)) + { + MemoryContextCallback *ctxcb; + + state = MemoryContextAllocZero(aggcontext, sizeof(MatrixSumState)); + state->cap = 16; + state->mats = MemoryContextAlloc(aggcontext, + state->cap * sizeof(GrB_Matrix)); + state->n = 0; + state->type = type; + + nargs = PG_NARGS(); + op = OS_GETARG_BINARYOP_HANDLE_OR_NULL(nargs, 2); + if (op == NULL) + op = plus_binaryop(type); + state->dup = op; + + ctxcb = MemoryContextAlloc(aggcontext, sizeof(MemoryContextCallback)); + ctxcb->func = context_callback_matrix_sum_free; + ctxcb->arg = state; + MemoryContextRegisterResetCallback(aggcontext, ctxcb); + } + else + { + state = (MatrixSumState *) PG_GETARG_POINTER(0); + } + + if (state->n == state->cap) + { + state->cap *= 2; + state->mats = repalloc(state->mats, + state->cap * sizeof(GrB_Matrix)); + } + + /* + * The input matrix lives in a per-row memory context that is reset before + * finalize, so we must dup it into a handle we own for the whole aggregate. + */ + OS_CHECK(GrB_Matrix_dup(&dupmat, A->matrix), + A->matrix, + "Cannot dup matrix in matrix_sum_matrix"); + state->mats[state->n++] = dupmat; + + PG_RETURN_POINTER(state); +} + +PG_FUNCTION_INFO_V1(matrix_sum_final); +Datum +matrix_sum_final(PG_FUNCTION_ARGS) +{ + MatrixSumState *state; + GrB_Matrix C; + char msg[LAGRAPH_MSG_LEN]; + + if (PG_ARGISNULL(0)) + PG_RETURN_NULL(); + + state = (MatrixSumState *) PG_GETARG_POINTER(0); + + if (state->n == 0) + PG_RETURN_NULL(); + + LA_CHECK(LAGraph_Matrix_Sum(&C, + state->mats, + (GrB_Index) state->n, + state->dup, + msg)); + + /* + * The dup'd inputs in state->mats are freed by the aggregate context reset + * callback; C is owned by the returned os_Matrix. + */ + OS_RETURN_MATRIX(new_matrix(NULL, 0, 0, CurrentMemoryContext, C)); +} + +#endif /* ONESPARSE_HAVE_LAGRAPH_MATRIX_SUM */ diff --git a/src/onesparse.c b/src/onesparse.c index becfa7a..5d1d992 100644 --- a/src/onesparse.c +++ b/src/onesparse.c @@ -292,6 +292,38 @@ default_binaryop(GrB_Type type) return NULL; } +/* + * The PLUS binary op for a builtin type. Used as the default duplicate-combiner + * (dup) for matrix_sum, so matrix_sum(m) computes the element-wise sum. + */ +GrB_BinaryOp +plus_binaryop(GrB_Type type) +{ + if (type == GrB_INT64) + return GrB_PLUS_INT64; + else if (type == GrB_INT32) + return GrB_PLUS_INT32; + else if (type == GrB_INT16) + return GrB_PLUS_INT16; + else if (type == GrB_INT8) + return GrB_PLUS_INT8; + if (type == GrB_UINT64) + return GrB_PLUS_UINT64; + else if (type == GrB_UINT32) + return GrB_PLUS_UINT32; + else if (type == GrB_UINT16) + return GrB_PLUS_UINT16; + else if (type == GrB_UINT8) + return GrB_PLUS_UINT8; + if (type == GrB_FP64) + return GrB_PLUS_FP64; + else if (type == GrB_FP32) + return GrB_PLUS_FP32; + else if (type == GrB_BOOL) + return GrB_LOR; + return NULL; +} + GrB_Monoid default_monoid(GrB_Type type) { diff --git a/src/onesparse.h b/src/onesparse.h index 05390fb..5fcfd6f 100644 --- a/src/onesparse.h +++ b/src/onesparse.h @@ -192,6 +192,7 @@ const char *error_name(GrB_Info info); GrB_Type type_promote(GrB_Type left, GrB_Type right); GrB_BinaryOp default_binaryop(GrB_Type type); +GrB_BinaryOp plus_binaryop(GrB_Type type); GrB_Monoid default_monoid(GrB_Type type); GrB_Semiring default_semiring(GrB_Type type); diff --git a/templates/matrix_header.sql b/templates/matrix_header.sql index 1165b38..3d7e454 100644 --- a/templates/matrix_header.sql +++ b/templates/matrix_header.sql @@ -394,21 +394,44 @@ RETURNS matrix AS '$libdir/onesparse', 'matrix_cast' LANGUAGE C STABLE; -CREATE FUNCTION matrix_agg_matrix(state matrix, a matrix) +-- matrix_agg_final is the finalfunc for the per-type element builder +-- matrix_agg(i, j, v) (see matrix_op.sql); leave it as-is. +CREATE FUNCTION matrix_agg_final(matrix) RETURNS matrix +AS '$libdir/onesparse', 'matrix_agg_final' +LANGUAGE C STRICT; + +-- The matrix-combining matrix_agg(a matrix) uses a binary-counter merge with an +-- internal carry-stack state (src/matrix/matrix_agg_matrix.c). The optional +-- binaryop selects the duplicate combiner; the default is PLUS, matching +-- matrix_sum and matrix_binary_sum. +CREATE FUNCTION matrix_agg_matrix(state internal, a matrix) +RETURNS internal AS '$libdir/onesparse', 'matrix_agg_matrix' -LANGUAGE C STABLE; +LANGUAGE C; -CREATE FUNCTION matrix_agg_final(matrix) +CREATE FUNCTION matrix_agg_matrix(state internal, a matrix, op binaryop) +RETURNS internal +AS '$libdir/onesparse', 'matrix_agg_matrix' +LANGUAGE C; + +CREATE FUNCTION matrix_agg_combine_final(state internal) RETURNS matrix -AS '$libdir/onesparse', 'matrix_agg_final' +AS '$libdir/onesparse', 'matrix_agg_combine_final' LANGUAGE C STRICT; CREATE AGGREGATE matrix_agg (a matrix) ( SFUNC=matrix_agg_matrix, - STYPE=matrix, - FINALFUNC=matrix_agg_final + STYPE=internal, + FINALFUNC=matrix_agg_combine_final + ); + +CREATE AGGREGATE matrix_agg (a matrix, op binaryop) + ( + SFUNC=matrix_agg_matrix, + STYPE=internal, + FINALFUNC=matrix_agg_combine_final ); CREATE FUNCTION set_element(a matrix, i bigint, j bigint, s scalar) diff --git a/templates/tests/test_matrix_header.sql b/templates/tests/test_matrix_header.sql index 2a6cf6e..eb9aa4d 100644 --- a/templates/tests/test_matrix_header.sql +++ b/templates/tests/test_matrix_header.sql @@ -769,6 +769,35 @@ drop table mat_agg_test; drop table mat_agg_float; drop table mat_agg_bool; +-- ## Combining matrices with `matrix_agg(a matrix)` +-- +-- The matrix-combining form of `matrix_agg` sums one matrix per row into a +-- single result using a binary-counter merge. The default duplicate combiner is +-- PLUS, so overlapping entries are added (matching `matrix_sum`). + +create temporary table mat_combine (a matrix); +insert into mat_combine values + ('int64(3:3)[0:0:1 1:1:2]'::matrix), + ('int64(3:3)[0:0:10 2:2:3]'::matrix), + ('int64(3:3)[0:0:100 1:1:20]'::matrix); + +-- Default (PLUS): (0,0)=111, (1,1)=22, (2,2)=3 +select print(matrix_agg(a)) as combined from mat_combine; + +-- The default equals an explicit PLUS, and matches the expected literal +select matrix_agg(a) = matrix_agg(a, 'plus_int64'::binaryop) as default_is_plus, + matrix_agg(a) = 'int64(3:3)[0:0:111 1:1:22 2:2:3]'::matrix as matches_expected +from mat_combine; + +-- A non-default combiner (MAX): (0,0)=100, (1,1)=20, (2,2)=3 +select matrix_agg(a, 'max_int64'::binaryop) = 'int64(3:3)[0:0:100 1:1:20 2:2:3]'::matrix as max_ok +from mat_combine; + +-- Empty input aggregates to NULL +select matrix_agg(a) is null as empty_is_null from mat_combine where false; + +drop table mat_combine; + -- ## Descriptor Basics -- -- Test basic descriptor construction and properties. diff --git a/test.sh b/test.sh index 9d85290..056b2d4 100755 --- a/test.sh +++ b/test.sh @@ -13,7 +13,17 @@ docker rm -f onesparse-test set -e echo building test image -docker build -f Dockerfile-debug --build-arg UID=$(id -u) --build-arg GID=$(id -g) . -t onesparse/test +# LAGRAPH_REF / LAGRAPH_REPO / WITH_MATRIX_SUM can be overridden in the +# environment to build against a LAGraph branch (e.g. the matrix-sum PR) and +# enable the experimental matrix_sum aggregate: +# LAGRAPH_REF=matrix-sum WITH_MATRIX_SUM=1 ./test.sh +docker build -f Dockerfile-debug \ + --build-arg UID=$(id -u) --build-arg GID=$(id -g) \ + --build-arg LAGRAPH_REPO="${LAGRAPH_REPO:-https://github.com/GraphBLAS/LAGraph.git}" \ + --build-arg LAGRAPH_REF="${LAGRAPH_REF:-v1.2.1}" \ + --build-arg WITH_MATRIX_SUM="${WITH_MATRIX_SUM:-0}" \ + --build-arg LAGRAPH_CACHE_BUST="${LAGRAPH_CACHE_BUST:-0}" \ + . -t onesparse/test docker run --user $(id -u):$(id -g) --mount type=bind,source=$(pwd),target=/home/postgres/onesparse,bind-propagation=rshared --cap-add=SYS_PTRACE --security-opt seccomp=unconfined -d --name "$DB_HOST" onesparse/test