view release on metacpan or  search on metacpan

xs/proto_native_parse.c  view on Meta::CPAN

                    goto data_error;
                }
                dpos++;

                /* Allocate LC dict state on first column of first block */
                if (c == 0 && !self->lc_dicts) {
                    Newxz(self->lc_dicts, num_cols, SV**);
                    Newxz(self->lc_dict_sizes, num_cols, uint64_t);
                    self->lc_num_cols = (int)num_cols;
                }

                {
                    int col_err = 0;
                    columns[c] = decode_column_ex(dbuf, dlen, &dpos, num_rows, col_types[c], &col_err, self->decode_flags, self, (int)c);
                    if (!columns[c]) {
                        if (col_err || decompressed) {
                            *errmsg = safe_strdup("decode_column failed");
                            goto data_error;
                        }
                        goto data_need_more;
                    }
                }
            }

            /* Convert column-oriented to row-oriented */
            {
            AV **target;
            if (ptype == SERVER_TOTALS) {
                if (!self->native_totals) self->native_totals = newAV();
                target = &self->native_totals;
            } else if (ptype == SERVER_EXTREMES) {
                if (!self->native_extremes) self->native_extremes = newAV();
                target = &self->native_extremes;
            } else {
                if (!self->native_rows) self->native_rows = newAV();
                target = &self->native_rows;
            }

            if (named) {
                for (r = 0; r < num_rows; r++) {
                    HV *hv = newHV();
                    for (c = 0; c < num_cols; c++) {
                        if (!hv_store(hv, cnames[c], cname_lens[c], columns[c][r], 0))
                            SvREFCNT_dec(columns[c][r]);
                    }
                    av_push(*target, newRV_noinc((SV*)hv));
                }
            } else {
                for (r = 0; r < num_rows; r++) {
                    AV *row = newAV();
                    if (num_cols > 0)
                        av_extend(row, num_cols - 1);
                    for (c = 0; c < num_cols; c++) {
                        av_push(row, columns[c][r]);
                    }
                    av_push(*target, newRV_noinc((SV*)row));
                }
            }
            }

            /* Fire on_data streaming callback if set (only for DATA, not TOTALS/EXTREMES) */
            {
                SV *on_data = (ptype == SERVER_DATA) ? peek_cb_on_data(self) : NULL;
                if (on_data && self->native_rows) {
                    /* Hold a reference across call_sv: a reentrant
                     * skip_pending() / cancel() in the handler would
                     * otherwise pop the cb_queue entry and free this
                     * callback while we're still invoking it. */
                    SvREFCNT_inc(on_data);
                    self->callback_depth++;
                    {
                        dSP;
                        ENTER; SAVETMPS;
                        PUSHMARK(SP);
                        PUSHs(sv_2mortal(newRV_inc((SV*)self->native_rows)));
                        PUTBACK;
                        call_sv(on_data, G_DISCARD | G_EVAL);
                        WARN_AND_CLEAR_ERRSV("on_data handler");
                        FREETMPS; LEAVE;
                    }
                    /* Decrement on_data BEFORE callback_depth-- so a DESTROY
                     * triggered by the dec (closure holding last $ch ref)
                     * still sees callback_depth > 0 and defers Safefree(self). */
                    SvREFCNT_dec(on_data);
                    self->callback_depth--;
                    /* Clear accumulated rows for next block */
                    CLEAR_SV(self->native_rows);
                    if (check_destroyed(self)) {
                        if (cnames) Safefree(cnames);
                        if (cname_lens) Safefree(cname_lens);
                        for (c = 0; c < num_cols; c++) {
                            Safefree(columns[c]);
                            free_col_type(col_types[c]);
                        }
                        Safefree(columns); Safefree(col_types);
                        if (decompressed) Safefree(decompressed);
                        return -2;
                    }
                }
            }

            /* Cleanup column arrays (SVs moved to rows, don't dec refcnt) */
            for (c = 0; c < num_cols; c++) {
                Safefree(columns[c]);
                free_col_type(col_types[c]);
            }
            Safefree(columns);
            Safefree(col_types);
            if (cnames) Safefree(cnames);
            if (cname_lens) Safefree(cname_lens);
            if (decompressed) Safefree(decompressed);
            else pos = dpos;  /* uncompressed: advance pos to match dpos */

            /* Consume from recv_buf */
            recv_consume(self, pos);
            return 1;

        data_error:
        data_need_more:
            /* Cleanup partial decode */
            for (c = 0; c < num_cols; c++) {