add: csv_read function

.gitignore

@@ -10,3 +10,4 @@ pgbench_log.*
 pg_csv--*.sql
 !pg_csv--*--*.sql
 tags
+bench/data/customers-1000000.csv

Makefile

@@ -38,6 +38,7 @@ REGRESS_OPTS = --inputdir=test
 
 MODULE_big = $(EXTENSION)
 SRC = $(wildcard $(SRC_DIR)/*.c)
+BENCH_DATA_DIR=bench/data
 
 ifdef BUILD_DIR
 OBJS = $(patsubst $(SRC_DIR)/%.c, $(BUILD_DIR)/%.o, $(SRC))
@@ -53,6 +54,8 @@ all: sql/$(EXTENSION)--$(EXTVERSION).sql $(EXTENSION).control
 
 build: $(BUILD_DIR)/$(EXTENSION).$(SHARED_EXT) sql/$(EXTENSION)--$(EXTVERSION).sql $(EXTENSION).control
 
+bench: $(BENCH_DATA_DIR)/customers-1000000.csv
+
 $(BUILD_DIR)/.gitignore: sql/$(EXTENSION)--$(EXTVERSION).sql $(EXTENSION).control
 	mkdir -p $(BUILD_DIR)/extension
 	cp $(EXTENSION).control $(BUILD_DIR)/extension
@@ -62,9 +65,17 @@ $(BUILD_DIR)/.gitignore: sql/$(EXTENSION)--$(EXTVERSION).sql $(EXTENSION).contro
 $(BUILD_DIR)/%.o: $(SRC_DIR)/%.c $(BUILD_DIR)/.gitignore
 	$(CC) $(CPPFLAGS) $(CFLAGS) -c $< -o $@
 
+$(BUILD_DIR)/pg_csv.o: $(SRC_DIR)/csv.h $(SRC_DIR)/cparsec.h
+src/pg_csv.o: $(SRC_DIR)/csv.h $(SRC_DIR)/cparsec.h
+
 $(BUILD_DIR)/$(EXTENSION).$(SHARED_EXT): $(EXTENSION).$(SHARED_EXT)
 	mv $? $@
 
+$(BENCH_DATA_DIR)/customers-1000000.csv: $(BENCH_DATA_DIR)/customers-1000000.7z
+	7z e -y $< -o$(BENCH_DATA_DIR) $(notdir $@)
+# needed to not decompress everytime, 7z preserves archive original timestamp and that trips up make
+	touch $@
+
 sql/$(EXTENSION)--$(EXTVERSION).sql: sql/$(EXTENSION).sql
 	cp $< $@
 

README.md

@@ -11,7 +11,7 @@ Postgres has CSV support on the [COPY](https://www.postgresql.org/docs/current/s
 
 `pg_csv` offers flexible CSV processing as a solution.
 
-- Includes a CSV aggregate that composes with SQL expressions.
+- Includes CSV export/import functions that compose with SQL expressions.
 - Native C extension, x2 times faster than SQL queries that try to output CSV
 - No dependencies except Postgres.
 
@@ -29,7 +29,9 @@ To install the extension:
 create extension pg_csv;
 ```
 
-## csv_agg
+## CSV Export
+
+### csv_agg
 
 Aggregate that builds a CSV respecting [RFC 4180](https://www.ietf.org/rfc/rfc4180.txt), quoting as required.
 
@@ -146,6 +148,46 @@ FROM   projects x;
 (1 row)
 ```
 
+## CSV Import
+
+### csv_read
+
+The `csv_read` function can read inline text values, respecting quoting and escaping inside quotes.
+
+```sql
+select id, name
+from csv_read(null::projects, E'1,IOS,4\n2,"Win""dows",4') where id = 2;
+ id |   name
+----+----------
+  2 | Win"dows
+```
+
+It can also be used to read from files inside the data directory.
+
+```sql
+select "First Name", "Company", "Website"
+from csv_read(
+  null::customers,
+  pg_read_file('data/customers-100.csv')
+)
+where "Index" = '4';
+ Index |           Customer Id           |         First Name         | Last Name | Company | City | Country | Phone 1 | Phone 2 | Email | Subscription Date | Website
+-------+---------------------------------+----------------------------+-----------+---------+------+---------+---------+---------+-------+-------------------+---------
+ Linda | Dominguez, Mcmillan and Donovan | http://www.good-lyons.com/ |           |         |      |         |         |         |       |                   |
+```
+
+This combines as usual with the `insert` statement.
+
+```sql
+insert into customers
+select *
+from csv_read(
+  null::customers,
+  pg_read_file('data/customers-100.csv')
+)
+limit 10; -- you can use LIMIT, OFFSET or WHERE
+```
+
 ## Limitations
 
 - For large bulk exports and imports, `COPY ... CSV` should still be preferred as it's more memory efficient due to streaming support.

bench/csv_read.sql

@@ -0,0 +1,8 @@
+truncate customers_csv;
+
+insert into customers_csv
+select *
+from csv_read(
+  null::customers_csv,
+  pg_read_file('data/customers-1000000.csv')
+);

bench/data/README.md

@@ -0,0 +1,7 @@
+The CSV samples were obtained from https://github.com/datablist/sample-csv-files.
+
+To reduce its size, the `customers-1000000.csv` example was compressed with:
+
+```
+7z a -t7z -m0=PPMd -mx=9 customers-1000000.7z customers-1000000.csv
+```

bench/init.sql

@@ -3,6 +3,21 @@
 
 create extension if not exists pg_csv;
 
+create unlogged table customers_csv (
+  "Index" text,
+  "Customer Id" text,
+  "First Name" text,
+  "Last Name" text,
+  "Company" text,
+  "City" text,
+  "Country" text,
+  "Phone 1" text,
+  "Phone 2" text,
+  "Email" text,
+  "Subscription Date" text,
+  "Website" text
+);
+
 CREATE TABLE customers (
   customer_id   CHAR(5) PRIMARY KEY,
   company_name  TEXT      NOT NULL,

bench/native_copy.sql

@@ -0,0 +1,6 @@
+truncate customers_csv;
+
+copy customers_csv from 'data/customers-1000000.csv' with (
+  format csv,
+  header true
+);

shell.nix

@@ -26,12 +26,20 @@ mkShellNoCC {
 
         pg_ver=$1
 
+        make bench > /dev/null
+
         for file in ./bench/*.sql; do
 
         if [ "$file" = ./bench/init.sql ]; then
           continue
         fi
 
+        duration=30
+        if [ "$file" = ./bench/native_copy.sql ] ||
+           [ "$file" = ./bench/csv_read.sql ]; then
+          duration=120
+        fi
+
         cat <<EOF
 
         ## "$file"
@@ -45,13 +53,14 @@ mkShellNoCC {
         results:
 
         \`\`\`
-        $(${xpg.xpg}/bin/xpg -v "$pg_ver" pgbench -n -c 1 -T 30 -M simple -f "$file")
+        $(${xpg.xpg}/bin/xpg -v "$pg_ver" pgbench -n -c 1 -T "$duration" -M simple -f "$file")
         \`\`\`
         EOF
 
         done
       '';
     in [
+      p7zip
       xpg.xpg
       style
       styleCheck

sql/pg_csv.sql

@@ -29,6 +29,11 @@ create function csv_agg_finalfn(internal)
   language c
 as 'MODULE_PATHNAME';
 
+create function csv_read(anyelement, text)
+  returns setof anyelement
+  language c
+as 'MODULE_PATHNAME';
+
 create aggregate csv_agg(anyelement) (
   sfunc     = csv_agg_transfn,
   stype     = internal,

src/csv.h

@@ -0,0 +1,52 @@
+#ifndef PG_CSV_PARSER_H
+#define PG_CSV_PARSER_H
+
+#define CPC_USE_STRING_H
+#define CPC_USE_UNNAMED
+#include "cparsec.h"
+
+// Strip the outer quotes and collapse doubled quotes inside one quoted field
+static inline CpcResult unescape_quoted(__attribute__((unused)) CpcArena *A, const CpcValue *v,
+                                        CpcSlice rest) {
+  // TODO should not happen
+  if (!v || v->kind != CPC_SLICE) return cpc_res_err(rest, "csv: expected slice", NULL);
+
+  CpcSlice s = v->as.slice;
+  // Leave non-quoted slices unchanged, only unescape fully quoted
+  if (s.len < 2 || s.ptr[0] != '"' || s.ptr[s.len - 1] != '"')
+    return cpc_res_ok(cpc_val_slice(s), rest);
+
+  char  *out = (char *)s.ptr + 1;
+  size_t dst = 0;
+  size_t src = 1;
+  size_t end = s.len - 1;
+  while (src < end) {
+    char c = s.ptr[src];
+    // Rewrite the quoted field in place, collapsing doubled quotes to one char
+    if (c == '"' && (src + 1) < end && s.ptr[src + 1] == '"') {
+      out[dst++] = '"';
+      src += 2;
+    } else {
+      out[dst++] = c;
+      src++;
+    }
+  }
+
+  return cpc_res_ok(cpc_val_slice((CpcSlice){.ptr = out, .len = dst}), rest);
+}
+
+extern CpcResult csvRow(CpcSlice input, CpcArena *A, const char *err);
+
+// semicolons are added just for to not make clang-format crazy
+static inline CPC_TAKE_QUOTED(quoted, '"', '"');
+static inline CPC_MAP(quotedField, quoted, unescape_quoted);
+static inline CPC_TAKE_TILL_ONE_OF(unquotedField, ",\r\n");
+static inline CPC_STRING(comma, ",");
+static inline CPC_ALT(field_, quotedField, unquotedField);
+static inline CPC_LABEL(field, field_, "field");
+static inline CPC_SEP_BY_1(record, field, comma);
+static inline CPC_ALT(lineEnd_, CPC_END_OF_LINE_, CPC_EOF_);
+static inline CPC_LABEL(lineEnd, lineEnd_, "expected newline or end of input");
+CPC_LEFT(csvRow, record, lineEnd);
+
+#endif