view release on metacpan or  search on metacpan

xs/proto_native_parse.c  view on Meta::CPAN

        }
    }

#undef _BAIL

    if (!decompressed) pos = bpos;
    if (decompressed) Safefree(decompressed);
    *outer_pos = pos;
    return 1;
}

/* Like parse_and_discard_block, but for each row of the block invokes
 * `cb` with one hashref keyed by column name. Used by on_log.
 * Caller has already verified cb is non-NULL. */
static int parse_and_emit_log_block(ev_clickhouse_t *self,
                                     const char *buf, size_t len, size_t *outer_pos,
                                     SV *cb, char **errmsg) {
    size_t pos = *outer_pos;
    int rc;
    char *decompressed = NULL;
    const char *bbuf;
    size_t blen, bpos;

    rc = skip_native_string(buf, len, &pos);
    if (rc == 0) return 0;
    if (rc < 0) { *errmsg = safe_strdup("malformed log block"); return -1; }

#ifdef HAVE_LZ4
    if (self->compress) {
        int need_more = 0;
        const char *lz4_err = NULL;
        decompressed = ch_lz4_decompress_chain(buf, len, &pos, &blen,
                                                &need_more, &lz4_err);
        if (!decompressed) {
            if (need_more) return 0;
            /* server log frames are not always compressed even with
             * compress=1 negotiated — fall through to raw parsing. */
        }
    }
    if (decompressed) { bbuf = decompressed; bpos = 0; }
    else
#endif
    { bbuf = buf; blen = len; bpos = pos; }

#define _BAIL_LOG(rc_val) do { \
    if (decompressed) Safefree(decompressed); \
    if (rc_val < 0) { *errmsg = safe_strdup("malformed log block"); } \
    return rc_val; \
} while (0)

    if (self->server_revision >= DBMS_MIN_REVISION_WITH_BLOCK_INFO) {
        rc = skip_block_info(bbuf, blen, &bpos);
        if (rc <= 0) _BAIL_LOG(rc);
    }
    uint64_t nc, nr;
    rc = read_varuint(bbuf, blen, &bpos, &nc);
    if (rc <= 0) _BAIL_LOG(rc);
    rc = read_varuint(bbuf, blen, &bpos, &nr);
    if (rc <= 0) _BAIL_LOG(rc);

    /* Collect column name + decoded values, then assemble per-row HVs. */
    char     **names = NULL;
    SV      ***data  = NULL;     /* data[col][row] */
    if (nc > 0) {
        /* A column occupies at least one wire byte (its name-length varint),
         * so more columns than remaining bytes is malformed — reject before
         * the allocation rather than letting Newxz attempt a huge size. */
        if (nc > (uint64_t)(blen - bpos)) _BAIL_LOG(-1);
        Newxz(names, nc, char *);
        Newxz(data,  nc, SV **);
    }
    /* err_seen: -1 = malformed, 0 = success / need-more, +1 = all columns
     * parsed cleanly. We distinguish "loop completed all nc columns" from
     * "loop broke early needing more data" via the explicit flag rather
     * than inspecting bpos, since bpos always advances past the header. */
    int err_seen = 1;
    for (uint64_t c = 0; c < nc; c++) {
        const char *cname; size_t cname_len;
        rc = read_native_string_ref(bbuf, blen, &bpos, &cname, &cname_len);
        if (rc <= 0) { err_seen = rc < 0 ? -1 : 0; break; }
        Newx(names[c], cname_len + 1, char);
        Copy(cname, names[c], cname_len, char);
        names[c][cname_len] = '\0';

        const char *ctype; size_t ctype_len;
        rc = read_native_string_ref(bbuf, blen, &bpos, &ctype, &ctype_len);
        if (rc <= 0) { err_seen = rc < 0 ? -1 : 0; break; }
        if (bpos >= blen) { err_seen = 0; break; }
        if ((uint8_t)bbuf[bpos]) { err_seen = -1; break; }
        bpos++;
        if (nr > 0) {
            col_type_t *ct = parse_col_type(ctype, ctype_len);
            int col_err = 0;
            SV **vals = decode_column(bbuf, blen, &bpos, nr, ct, &col_err, 0);
            free_col_type(ct);
            if (!vals) { err_seen = col_err ? -1 : 0; break; }
            data[c] = vals;
        }
    }

    if (err_seen == 1) {
        /* Pin cb across the loop: an on_log handler that calls
         * $ch->on_log(undef) would otherwise free the CV mid-iteration. */
        SvREFCNT_inc(cb);
        self->callback_depth++;
        for (uint64_t r = 0; r < nr; r++) {
            HV *row = newHV();
            for (uint64_t c = 0; c < nc; c++) {
                SV *v = data[c][r];
                SvREFCNT_inc(v);    /* hv_store consumes one ref */
                (void)hv_store(row, names[c], strlen(names[c]), v, 0);
            }
            dSP;
            ENTER; SAVETMPS; PUSHMARK(SP);
            EXTEND(SP, 1);
            PUSHs(sv_2mortal(newRV_noinc((SV *)row)));
            PUTBACK;
            call_sv(cb, G_EVAL | G_VOID | G_DISCARD);
            WARN_AND_CLEAR_ERRSV("on_log");
            FREETMPS; LEAVE;
            /* freed state is authoritatively handled by check_destroyed below */