feat: improve DX of csv_agg with csv_options

Now we use the following interface for the agg:

```sql
SELECT csv_agg(x, csv_options(';'))
SELECT csv_agg(x, csv_options(delimiter := '|'))

SELECT csv_agg(x) -- works as usual
```

This using a composite type plus a function constructor.

Also bumps version to 0.2.

Author: steve-chavez

.gitignore

@@ -8,3 +8,4 @@ results/
 pgbench_log.*
 .history
 pg_csv--*.sql
+!pg_csv--*--*.sql

Makefile

@@ -26,7 +26,7 @@ else
 endif
 
 EXTENSION = pg_csv
-EXTVERSION = 0.1
+EXTVERSION = 0.2
 
 DATA = $(wildcard sql/*--*.sql)
 

bench/csv_agg_delim.sql

@@ -1,5 +1,5 @@
 \set lim random(1000, 2000)
 
-select csv_agg(t,'|') from (
+select csv_agg(t, csv_options('|')) from (
   select * from student_emotion_assessments limit :lim
 ) as t;

sql/pg_csv--0.1--0.2.sql

@@ -0,0 +1,23 @@
+drop aggregate if exists csv_agg (anyelement, "char");
+drop function  if exists csv_agg_transfn (internal, anyelement, "char");
+
+create type csv_options as (
+  delimiter "char"
+);
+
+create function csv_options(delimiter "char" default ',') returns csv_options as $$
+  select row(delimiter)::csv_options;
+$$ language sql;
+
+create function csv_agg_transfn(internal, anyelement, csv_options)
+  returns internal
+  language c
+  as 'pg_csv';
+
+create aggregate csv_agg(anyelement, csv_options) (
+  sfunc     = csv_agg_transfn,
+  stype     = internal,
+  finalfunc = csv_agg_finalfn,
+  parallel  = safe
+);
+

sql/pg_csv--0.1.sql

@@ -0,0 +1,28 @@
+create function csv_agg_transfn(internal, anyelement)
+  returns internal
+  language c
+  as 'pg_csv';
+
+create function csv_agg_transfn(internal, anyelement, "char")
+  returns internal
+  language c
+  as 'pg_csv';
+
+create function csv_agg_finalfn(internal)
+  returns text
+  language c
+  as 'pg_csv';
+
+create aggregate csv_agg(anyelement, "char") (
+  sfunc     = csv_agg_transfn,
+  stype     = internal,
+  finalfunc = csv_agg_finalfn,
+  parallel  = safe
+);
+
+create aggregate csv_agg(anyelement) (
+  sfunc     = csv_agg_transfn,
+  stype     = internal,
+  finalfunc = csv_agg_finalfn,
+  parallel  = safe
+);

sql/pg_csv.sql

@@ -1,9 +1,17 @@
+create type csv_options as (
+  delimiter "char"
+);
+
+create function csv_options(delimiter "char" default ',') returns csv_options as $$
+  select row(delimiter)::csv_options;
+$$ language sql;
+
 create function csv_agg_transfn(internal, anyelement)
   returns internal
   language c
   as 'pg_csv';
 
-create function csv_agg_transfn(internal, anyelement, "char")
+create function csv_agg_transfn(internal, anyelement, csv_options)
   returns internal
   language c
   as 'pg_csv';
@@ -13,16 +21,17 @@ create function csv_agg_finalfn(internal)
   language c
   as 'pg_csv';
 
-create aggregate csv_agg(anyelement, "char") (
+create aggregate csv_agg(anyelement) (
   sfunc     = csv_agg_transfn,
   stype     = internal,
   finalfunc = csv_agg_finalfn,
   parallel  = safe
 );
 
-create aggregate csv_agg(anyelement) (
+create aggregate csv_agg(anyelement, csv_options) (
   sfunc     = csv_agg_transfn,
   stype     = internal,
   finalfunc = csv_agg_finalfn,
   parallel  = safe
 );
+

src/pg_csv.c

@@ -6,11 +6,16 @@ static const char NEWLINE = '\n';
 static const char DQUOTE  = '"';
 static const char CR      = '\r';
 
+typedef struct {
+  char delim;
+} CsvOptions;
+
 typedef struct {
   StringInfoData accum_buf;
   bool           header_done;
   bool           first_row;
   TupleDesc      tupdesc;
+  CsvOptions    *options;
 } CsvAggState;
 
 static inline bool is_reserved(char c) {
@@ -48,40 +53,65 @@ static char *datum_to_cstring(Datum datum, Oid typeoid) {
   return OidOutputFunctionCall(out_func, datum);
 }
 
+static void parse_csv_options(HeapTupleHeader opts_hdr, CsvOptions *csv_opts) {
+  // defaults
+  csv_opts->delim = ',';
+
+  if (opts_hdr == NULL) return;
+
+  TupleDesc desc = lookup_rowtype_tupdesc(HeapTupleHeaderGetTypeId(opts_hdr),
+                                          HeapTupleHeaderGetTypMod(opts_hdr));
+
+  Datum values[1];
+  bool  nulls[1];
+
+  heap_deform_tuple(
+      &(HeapTupleData){.t_len = HeapTupleHeaderGetDatumLength(opts_hdr), .t_data = opts_hdr}, desc,
+      values, nulls);
+
+  if (!nulls[0]) {
+    csv_opts->delim = DatumGetChar(values[0]);
+    if (is_reserved(csv_opts->delim))
+      ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+                      errmsg("delimiter cannot be newline, carriage return or "
+                             "double quote")));
+  }
+
+  ReleaseTupleDesc(desc);
+}
+
 PG_FUNCTION_INFO_V1(csv_agg_transfn);
 Datum csv_agg_transfn(PG_FUNCTION_ARGS) {
-  CsvAggState *state;
+  CsvAggState    *state = !PG_ARGISNULL(0) ? (CsvAggState *)PG_GETARG_POINTER(0) : NULL;
+  HeapTupleHeader next  = !PG_ARGISNULL(1) ? PG_GETARG_HEAPTUPLEHEADER(1) : NULL;
 
   // first call when the accumulator is NULL
   // pretty standard stuff, for example see the jsonb_agg transition function
   // https://github.com/postgres/postgres/blob/3c4e26a62c31ebe296e3aedb13ac51a7a35103bd/src/backend/utils/adt/jsonb.c#L1521
-  if (PG_ARGISNULL(0)) {
+  if (state == NULL) {
     MemoryContext aggctx, oldctx;
 
     if (!AggCheckCallContext(fcinfo, &aggctx))
       elog(ERROR, "csv_agg_transfn called in non‑aggregate context");
 
     oldctx = MemoryContextSwitchTo(aggctx);
 
-    state = (CsvAggState *)palloc(sizeof(CsvAggState));
+    state = palloc(sizeof(CsvAggState));
     initStringInfo(&state->accum_buf);
     state->header_done = false;
     state->first_row   = true;
     state->tupdesc     = NULL;
+    state->options     = palloc(sizeof(CsvOptions));
 
-    MemoryContextSwitchTo(oldctx);
-  } else
-    state = (CsvAggState *)PG_GETARG_POINTER(0);
-
-  if (PG_ARGISNULL(1)) PG_RETURN_POINTER(state); // skip NULL rows
-
-  HeapTupleHeader next = PG_GETARG_HEAPTUPLEHEADER(1);
+    // we'll parse the csv options only once
+    HeapTupleHeader opts_hdr =
+        PG_NARGS() >= 3 && !PG_ARGISNULL(2) ? PG_GETARG_HEAPTUPLEHEADER(2) : NULL;
+    parse_csv_options(opts_hdr, state->options);
 
-  char delim = PG_NARGS() >= 3 && !PG_ARGISNULL(2) ? PG_GETARG_CHAR(2) : ',';
+    MemoryContextSwitchTo(oldctx);
+  }
 
-  if (is_reserved(delim))
-    ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-                    errmsg("delimiter cannot be newline, carriage return or double quote")));
+  if (next == NULL) PG_RETURN_POINTER(state); // skip NULL rows
 
   // build header and cache tupdesc once
   if (!state->header_done) {
@@ -95,10 +125,10 @@ Datum csv_agg_transfn(PG_FUNCTION_ARGS) {
         continue;
 
       if (i > 0) // only append delimiter after the first value
-        appendStringInfoChar(&state->accum_buf, delim);
+        appendStringInfoChar(&state->accum_buf, state->options->delim);
 
       char *cstr = NameStr(att->attname);
-      csv_append_field(&state->accum_buf, cstr, strlen(cstr), delim);
+      csv_append_field(&state->accum_buf, cstr, strlen(cstr), state->options->delim);
     }
 
     appendStringInfoChar(&state->accum_buf, NEWLINE);
@@ -131,12 +161,12 @@ Datum csv_agg_transfn(PG_FUNCTION_ARGS) {
     if (att->attisdropped) // pg always keeps dropped columns, guard against this
       continue;
 
-    if (i > 0) appendStringInfoChar(&state->accum_buf, delim);
+    if (i > 0) appendStringInfoChar(&state->accum_buf, state->options->delim);
 
     if (nulls[i]) continue; // empty field for NULL
 
     char *cstr = datum_to_cstring(datums[i], att->atttypid);
-    csv_append_field(&state->accum_buf, cstr, strlen(cstr), delim);
+    csv_append_field(&state->accum_buf, cstr, strlen(cstr), state->options->delim);
   }
 
   PG_RETURN_POINTER(state);

test/expected/delimiters.out

@@ -1,5 +1,5 @@
 -- semicolon delimiter
-SELECT csv_agg(x, ';') AS body
+SELECT csv_agg(x, csv_options(';')) AS body
 FROM   projects x;
              body              
 -------------------------------
@@ -17,8 +17,8 @@ FROM   projects x;
   CRLF""";8
 (1 row)
 
--- pipe delimiter
-SELECT csv_agg(x, '|') AS body
+-- pipe delimiter, named params work too
+SELECT csv_agg(x, csv_options(delimiter := '|')) AS body
 FROM   projects x;
              body              
 -------------------------------
@@ -37,7 +37,7 @@ FROM   projects x;
 (1 row)
 
 -- tab delimiter
-SELECT csv_agg(x, E'\t') AS body
+SELECT csv_agg(x, csv_options(E'\t')) AS body
 FROM   projects x;
                    body                    
 -------------------------------------------
@@ -56,14 +56,14 @@ FROM   projects x;
 (1 row)
 
 -- newline is forbidden as delimiter
-SELECT csv_agg(x, E'\n') AS body
+SELECT csv_agg(x, csv_options(E'\n')) AS body
 FROM   projects x;
 ERROR:  delimiter cannot be newline, carriage return or double quote
 -- double quote is forbidden as delimiter
-SELECT csv_agg(x, '"') AS body
+SELECT csv_agg(x, csv_options('"')) AS body
 FROM   projects x;
 ERROR:  delimiter cannot be newline, carriage return or double quote
 -- carriage return is forbidden as delimiter
-SELECT csv_agg(x, E'\r') AS body
+SELECT csv_agg(x, csv_options(E'\r')) AS body
 FROM   projects x;
 ERROR:  delimiter cannot be newline, carriage return or double quote

test/sql/delimiters.sql

@@ -1,23 +1,23 @@
 -- semicolon delimiter
-SELECT csv_agg(x, ';') AS body
+SELECT csv_agg(x, csv_options(';')) AS body
 FROM   projects x;
 
--- pipe delimiter
-SELECT csv_agg(x, '|') AS body
+-- pipe delimiter, named params work too
+SELECT csv_agg(x, csv_options(delimiter := '|')) AS body
 FROM   projects x;
 
 -- tab delimiter
-SELECT csv_agg(x, E'\t') AS body
+SELECT csv_agg(x, csv_options(E'\t')) AS body
 FROM   projects x;
 
 -- newline is forbidden as delimiter
-SELECT csv_agg(x, E'\n') AS body
+SELECT csv_agg(x, csv_options(E'\n')) AS body
 FROM   projects x;
 
 -- double quote is forbidden as delimiter
-SELECT csv_agg(x, '"') AS body
+SELECT csv_agg(x, csv_options('"')) AS body
 FROM   projects x;
 
 -- carriage return is forbidden as delimiter
-SELECT csv_agg(x, E'\r') AS body
+SELECT csv_agg(x, csv_options(E'\r')) AS body
 FROM   projects x;