view release on metacpan or  search on metacpan

xs/proto_native_parse.c  view on Meta::CPAN

        if (rc == 0) {
            /* need more data */
            return;
        }

        if (rc == -2) {
            /* object destroyed inside callback */
            return;
        }

        if (rc == 4) {
            /* ServerHello received — send addendum (revision >= 54458) */
            if (self->native_state == NATIVE_WAIT_HELLO) {
                /* Addendum: quota_key (only if server supports it) */
                if (self->server_revision >= DBMS_MIN_PROTOCOL_VERSION_WITH_ADDENDUM) {
                    native_buf_t ab;
                    nbuf_init(&ab);
                    nbuf_cstring(&ab, "");  /* quota_key */
                    send_replace(self, ab.data, ab.len);
                    if (try_write(self)) return;
                    if (self->send_pos < self->send_len) {
                        /* Addendum partially written (EAGAIN); io_cb finishes
                         * connect once the buffer drains, otherwise pipeline_advance
                         * could overwrite the unsent tail with a queued query. */
                        self->pending_addendum_finish = 1;
                        return;
                    }
                }
                self->native_state = NATIVE_IDLE;
                if (finish_connect(self)) return;
            }
            /* pipeline_advance -> try_write may free self; no data
             * in recv_buf for the just-dispatched request yet */
            return;
        }

        if (rc == -1) {
            /* Stream corruption: native packets aren't length-delimited,
             * so resync is impossible — the server's remaining output
             * for this query would be misattributed to the next one.
             * Tear down; auto_reconnect re-establishes an idle link. */
            if (self->native_state == NATIVE_WAIT_HELLO) {
                /* Skip auto_reconnect on a malformed ServerHello: the peer
                 * isn't a ClickHouse server, so retrying just spins.
                 * (connect_timeout still uses fail_connection, which retries.) */
                teardown_io_error(self, errmsg, "connection failed");
                Safefree(errmsg);
                return;
            }
            fail_connection(self, errmsg);
            Safefree(errmsg);
            return;
        }

        if (rc == 5) {
            /* Server exception — a normal query error with its bytes
             * consumed, so the stream stays in sync: deliver to the
             * query callback and pipeline the next query. */
            if (self->native_state == NATIVE_WAIT_HELLO) {
                /* The server rejected the handshake itself (e.g. wrong
                 * credentials) — fail fast with the real text, to
                 * on_error and to any pre-connect queued queries. No
                 * reconnect: retrying a rejected handshake just spins. */
                teardown_io_error(self, errmsg, errmsg);
                Safefree(errmsg);
                return;
            }
            stop_timing(self);

            CLEAR_SV(self->native_rows);
            CLEAR_INSERT(self);
            CLEAR_STR(self->insert_err);
            self->native_state = NATIVE_IDLE;
            self->cancel_after_meta = 0;
            self->cancelled = 0;
            self->recv_len = 0; /* drop this dead query's tail, if coalesced */
            if (self->send_count > 0) self->send_count--;
            lc_free_dicts(self);
            int destroyed = deliver_error(self, errmsg);
            Safefree(errmsg);
            if (destroyed) return;

            /* advance pipeline — may free self via try_write error */
            pipeline_advance(self);
            return;
        }

        if (rc == 2) {
            /* EndOfStream — deliver accumulated rows or deferred error */
            stop_timing(self);
            self->native_state = NATIVE_IDLE;
            self->cancel_after_meta = 0;
            CLEAR_INSERT(self);
            if (self->send_count > 0) self->send_count--;
            lc_free_dicts(self);

            /* Flush any uncoalesced progress accumulated since the last
             * fire so users instrumenting via on_progress see the full
             * total when the query completes within one progress_period
             * of the last fire. */
            if (self->on_progress && self->progress_period > 0) {
                int any = 0, i;
                for (i = 0; i < 5; i++) if (self->progress_acc[i]) { any = 1; break; }
                if (any) {
                    uint64_t pp[5];
                    memcpy(pp, self->progress_acc, sizeof(pp));
                    memset(self->progress_acc, 0, sizeof(self->progress_acc));
                    if (fire_progress_cb(self, pp) < 0) return;
                }
            }

            if (self->insert_err) {
                char *err = self->insert_err;
                self->insert_err = NULL;
                CLEAR_SV(self->native_rows);
                int destroyed = deliver_error(self, err);
                Safefree(err);
                if (destroyed) return;
            } else if (self->cancelled) {
                self->cancelled = 0;
                CLEAR_SV(self->native_rows);