use native format for http

CHANGELOG.md

@@ -84,6 +84,11 @@ All notable changes to this project will be documented in this file. It uses the
 *   Coerce array elements in binary driver. `Array(Int32)` to `bigint[]`, or
     `quantilesExactLow()` results into `double precision[]`, no longer fails
     with `could not cast value from integer[] to bigint[]` ([#326]).
+*   HTTP driver now uses ClickHouse's Native format, sharing encode/decode with
+    binary driver. `clickhouse_raw_query()` keeps TabSeparated behavior.
+*   Deprecated `fetch_size`. It is ignored and produces a warning. HTTP driver
+    always streams, handing Native decoder one chunk at a time. Decoded memory
+    is set by block size ClickHouse writes, not by client-side byte count.
 
 ### 🐞 Bug Fixes
 

Makefile

@@ -33,7 +33,7 @@ CH_C_DIR = $(PGCH_DIR)/clickhouse-c
 PG_CPPFLAGS = -I./src/include -isystem $(CH_C_DIR) -isystem $(PGCH_DIR) -isystem $(shell $(PG_CONFIG) --includedir-server) -DPGCH_MSG_PREFIX='"pg_clickhouse: "'
 
 # Link OpenSSL (for TLS in the binary driver), curl (for the HTTP driver),
-# libuuid (for http_streaming.c's query-id generator), and lz4 / zstd
+# libuuid (for http.c's query-id generator), and lz4 / zstd
 # (for the binary driver's compressed-frame codecs).
 PG_LDFLAGS = -lssl -lcrypto -llz4 -lzstd $(shell $(CURL_CONFIG) --libs)
 

doc/pg_clickhouse.md

@@ -11,7 +11,7 @@ CREATE EXTENSION pg_clickhouse;
 
 This library contains PostgreSQL extension that enables remote query execution
 on ClickHouse databases, including a [foreign data wrapper]. It supports
-PostgreSQL 13 and higher and ClickHouse 23 and higher.
+PostgreSQL 13 and higher and ClickHouse 23.3 and higher.
 
 ## Getting Started
 
@@ -142,10 +142,6 @@ The supported options are:
     "none", "lz4", or "zstd". Defaults to "lz4". Ignored by the "http" driver.
 *   `dbname`: The ClickHouse database to use upon connecting. Defaults to
     "default".
-*   `fetch_size`: Approximate batch size in bytes for HTTP streaming. Batches
-    split on row boundaries. Defaults to `50000000` (50 MB). `0` disables
-    streaming and buffers the full response. Foreign tables can override this
-    value.
 *   `host`: The host name of the ClickHouse server. Defaults to "localhost";
 *   `min_tls_version`: Minimum TLS protocol version to negotiate on connections
     that use TLS. One of `TLSv1`, `TLSv1.1`, `TLSv1.2`, or `TLSv1.3`. Defaults
@@ -315,9 +311,6 @@ The supported table options are:
 
 *   `database`: The name of the remote database. Defaults to the database
     defined for the foreign server.
-*   `fetch_size`: Approximate batch size in bytes for HTTP streaming.
-    Overrides server-level `fetch_size`. Defaults to `50000000` (50 MB). `0`
-    disables streaming and buffers the full response.
 *   `table_name`: The name of the remote table. Default to the name specified
     for the foreign table.
 *   `engine`: The [table engine] used by the ClickHouse table. For

src/cursor.c

@@ -0,0 +1,338 @@
+/*
+ * cursor.c
+ *
+ * Shared lifecycle for a cursor over a ClickHouse Native block stream: open
+ * over a driver's response, fetch rows through the decoder, release both on
+ * teardown. API lives in src/include/cursor.h; the drivers in src/pglink.c
+ * only supply the response and how to read it.
+ */
+
+#include "postgres.h"
+
+#include "catalog/pg_type_d.h"
+#include "miscadmin.h"
+#include "utils/builtins.h"
+#include "utils/memutils.h"
+#include "utils/portal.h"
+
+#include "cursor.h"
+
+/* Release the reader before the response its blocks came from. */
+static void
+cursor_free(void* c) {
+    ch_cursor* cursor = c;
+
+    pgch_reader_free(&cursor->reader);
+    cursor->free_response(cursor->response);
+    cursor->response = NULL;
+}
+
+/* Report decoder error; a driver hook may report cancellation instead. */
+static void
+raise_reader_error(ch_cursor* cursor) {
+    if (cursor->raise_response_error) {
+        cursor->raise_response_error(cursor);
+    }
+    /* Prefer consistent interrupt error message when fetch interrupted */
+    CHECK_FOR_INTERRUPTS();
+    ereport(
+        ERROR,
+        errcode(ERRCODE_SQL_ROUTINE_EXCEPTION),
+        errmsg("pg_clickhouse: %s", cursor->reader.error),
+        errdetail_internal("Remote Query: %.64000s", cursor->query)
+    );
+}
+
+/* Match the returned columns against the destination the query asked for. */
+static void
+configure_columns(ch_cursor* cursor, const ch_query* query) {
+    pgch_reader* reader = &cursor->reader;
+
+    if (query->tupdesc && query->attr_nums && cursor->columns_count > 0 &&
+        (size_t)list_length(query->attr_nums) != cursor->columns_count) {
+        ereport(
+            ERROR,
+            errcode(ERRCODE_DATATYPE_MISMATCH),
+            errmsg_internal(
+                "pg_clickhouse: returned %lu columns, expected %lu",
+                (unsigned long)cursor->columns_count,
+                (unsigned long)list_length(query->attr_nums)
+            ),
+            errdetail_internal("Remote Query: %.64000s", query->sql)
+        );
+    }
+
+    /* Preserve JSON text when PostgreSQL destination uses json, not jsonb. */
+    if (query->tupdesc && reader->coltypes) {
+        ListCell* lc;
+        size_t j = 0;
+
+        foreach (lc, query->attr_nums) {
+            int i = lfirst_int(lc);
+
+            if (reader->coltypes[j] == JSONBOID &&
+                TupleDescAttr(query->tupdesc, i - 1)->atttypid == JSONOID) {
+                reader->coltypes[j] = JSONOID;
+            }
+            j++;
+        }
+    }
+}
+
+ch_cursor*
+chfdw_cursor_open(void* conn, const ch_query* query, const ch_cursor_source* src) {
+    /* volatile: assigned inside PG_TRY, read after longjmp in PG_CATCH */
+    ch_cursor* volatile cursor  = NULL;
+    volatile MemoryContext cxt  = NULL;
+    volatile bool owns_response = false;
+    MemoryContext oldcxt        = CurrentMemoryContext;
+
+    PG_TRY();
+    {
+        cxt = AllocSetContextCreate(
+            PortalContext, "pg_clickhouse cursor", ALLOCSET_DEFAULT_SIZES
+        );
+        MemoryContextSwitchTo(cxt);
+
+        cursor                       = palloc0(sizeof(ch_cursor));
+        cursor->memcxt               = cxt;
+        cursor->conn                 = conn;
+        cursor->query                = pstrdup(query->sql);
+        cursor->free_response        = src->free_response;
+        cursor->raise_response_error = src->raise_response_error;
+
+        /* Register before taking the response, so unwinding releases it. */
+        cursor->callback.func = cursor_free;
+        cursor->callback.arg  = (void*)cursor;
+        MemoryContextRegisterResetCallback(cxt, &cursor->callback);
+        cursor->response = src->response;
+        owns_response    = true;
+
+        /* Blocks decode into cxt, outliving the per-row context. */
+        src->init_reader(&cursor->reader, cursor->response);
+        cursor->columns_count = pgch_reader_columns(&cursor->reader);
+
+        if (cursor->reader.error) {
+            raise_reader_error(cursor);
+        }
+        configure_columns(cursor, query);
+
+        MemoryContextSwitchTo(oldcxt);
+    }
+    PG_CATCH();
+    {
+        MemoryContextSwitchTo(oldcxt);
+        if (!owns_response) {
+            src->free_response(src->response);
+        }
+        if (cxt != NULL) {
+            MemoryContextDelete(cxt);
+        }
+        PG_RE_THROW();
+    }
+    PG_END_TRY();
+
+    return cursor;
+}
+
+/* Conversion state and target attribute per returned column. */
+static void
+build_conversion(ch_cursor* cursor, const ChFdwScanRowContext* ctx) {
+    pgch_reader* reader = &cursor->reader;
+    MemoryContext old   = MemoryContextSwitchTo(cursor->memcxt);
+    size_t ncols        = pgch_reader_columns(reader);
+    ListCell* lc;
+    size_t j = 0;
+
+    cursor->conversion_states = palloc0(ncols * sizeof(void*));
+    cursor->fill_dest         = palloc0(ncols * sizeof(int));
+    foreach (lc, ctx->retrieved_attrs) {
+        int attnum            = lfirst_int(lc);
+        Form_pg_attribute att = TupleDescAttr(ctx->tupdesc, attnum - 1);
+
+        cursor->fill_dest[j] = attnum - 1;
+        cursor->conversion_states[j] =
+            pgch_reader_convert_init(reader, j, att->atttypid, att->atttypmod);
+        j++;
+    }
+
+    MemoryContextSwitchTo(old);
+}
+
+/* Apply PostgreSQL conversions to fetched Native row. */
+static Datum*
+apply_row(ChFdwScanRowContext* ctx) {
+    ch_cursor* cursor   = ctx->cursor;
+    List* attrs         = ctx->retrieved_attrs;
+    TupleDesc tupdesc   = ctx->tupdesc;
+    Datum* values       = ctx->values;
+    bool* nulls         = ctx->nulls;
+    pgch_reader* reader = &cursor->reader;
+    size_t attcount     = list_length(attrs);
+
+    if (attcount == 0) {
+        if (pgch_reader_columns(reader) == 1 && reader->nulls[0]) {
+            nulls[0] = true;
+            return reader->values;
+        }
+        ereport(
+            ERROR,
+            errcode(ERRCODE_FDW_ERROR),
+            errmsg(
+                "pg_clickhouse: unexpected state: attributes "
+                "count == 0 and haven't got NULL in the response"
+            )
+        );
+    } else if (attcount != pgch_reader_columns(reader)) {
+        ereport(
+            ERROR,
+            errcode(ERRCODE_DATATYPE_MISMATCH),
+            errmsg_internal(
+                "pg_clickhouse: returned %lu columns, expected %lu",
+                pgch_reader_columns(reader),
+                attcount
+            )
+        );
+    }
+
+    if (tupdesc) {
+        Assert(values && nulls);
+
+        if (cursor->conversion_states == NULL) {
+            build_conversion(cursor, ctx);
+        }
+        pgch_reader_fill_map(
+            reader, cursor->conversion_states, cursor->fill_dest, values, nulls
+        );
+    }
+
+    return reader->values;
+}
+
+static void
+fetch_row_errcb(void* arg) {
+    const char* sql = (const char*)arg;
+
+    errdetail_internal("Remote Query: %.64000s", sql);
+}
+
+/*
+ * Fetch a row from the cursor and return its values.
+ *
+ * If ctx->tupdesc is set, ctx->attinmeta must also be set, and ctx->values
+ * and ctx->nulls must already be palloc'd with space for ctx->tupdesc->natts
+ * values.
+ *
+ * Use ctx->tupdesc and ctx->attinmeta to convert the values to the
+ * appropriate Datums, and store them and the indication of their NULLness in
+ * ctx->values and ctx->nulls, respectively, then return ctx->values.
+ *
+ * If ctx->tupdesc is not set, treat all values as text and return them as
+ * text `Datum`s. This is the use case for `chfdw_construct_create_tables()`,
+ * which only cares about text.
+ */
+Datum*
+chfdw_cursor_fetch_row(ChFdwScanRowContext* ctx) {
+    ch_cursor* cursor = ctx->cursor;
+    ErrorContextCallback errcallback;
+    bool have_data;
+    Datum* result;
+
+    errcallback.callback = fetch_row_errcb;
+    errcallback.arg      = (void*)cursor->query;
+    errcallback.previous = error_context_stack;
+    error_context_stack  = &errcallback;
+
+    have_data = pgch_reader_next(&cursor->reader);
+
+    if (cursor->reader.error) {
+        error_context_stack = errcallback.previous;
+        raise_reader_error(cursor);
+    }
+
+    result = have_data ? apply_row(ctx) : NULL;
+
+    error_context_stack = errcallback.previous;
+    return result;
+}
+
+/*
+ * Escape characters that would otherwise corrupt the tab/newline framing or
+ * collide with the \N null marker. Matches CH's TabSeparated escaping; \0 is
+ * unreachable since values arrive as cstrings, so it needs no case.
+ */
+static void
+append_tsv_escaped(StringInfo buf, const char* s) {
+    for (; *s != '\0'; s++) {
+        switch (*s) {
+        case '\\':
+            appendStringInfoString(buf, "\\\\");
+            break;
+        case '\b':
+            appendStringInfoString(buf, "\\b");
+            break;
+        case '\f':
+            appendStringInfoString(buf, "\\f");
+            break;
+        case '\n':
+            appendStringInfoString(buf, "\\n");
+            break;
+        case '\r':
+            appendStringInfoString(buf, "\\r");
+            break;
+        case '\t':
+            appendStringInfoString(buf, "\\t");
+            break;
+        default:
+            appendStringInfoChar(buf, *s);
+        }
+    }
+}
+
+text*
+chfdw_cursor_render_tsv(ch_cursor* cursor) {
+    pgch_reader* reader = &cursor->reader;
+    size_t ncols        = pgch_reader_columns(reader);
+    StringInfoData buf;
+
+    if (ncols == 0) {
+        return NULL;
+    }
+
+    initStringInfo(&buf);
+
+    while (pgch_reader_next(reader)) {
+        for (size_t i = 0; i < ncols; i++) {
+            if (i > 0) {
+                appendStringInfoChar(&buf, '\t');
+            }
+
+            if (reader->nulls[i]) {
+                appendStringInfoString(&buf, "\\N");
+            } else {
+                char* val =
+                    pgch_value_to_cstring(reader->coltypes[i], reader->values[i]);
+
+                append_tsv_escaped(&buf, val);
+                pfree(val);
+            }
+        }
+        appendStringInfoChar(&buf, '\n');
+        CHECK_FOR_INTERRUPTS();
+    }
+
+    if (reader->error) {
+        ereport(
+            ERROR,
+            errcode(ERRCODE_SQL_ROUTINE_EXCEPTION),
+            errmsg("pg_clickhouse: %s", reader->error)
+        );
+    }
+
+    if (buf.len == 0) {
+        pfree(buf.data);
+        return NULL;
+    }
+
+    return cstring_to_text_with_len(buf.data, buf.len);
+}