cursor.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (c) 2012-2019 Snowflake Computing Inc. All right reserved.
#
import logging
import re
import signal
import sys
import uuid
from logging import getLogger
from threading import Lock, Timer
from typing import Optional

from .compat import BASE_EXCEPTION_CLASS
from .constants import FIELD_NAME_TO_ID, PARAMETER_PYTHON_CONNECTOR_QUERY_RESULT_FORMAT
from .errorcode import (
    ER_CURSOR_IS_CLOSED,
    ER_FAILED_TO_REWRITE_MULTI_ROW_INSERT,
    ER_INVALID_VALUE,
    ER_NO_ARROW_RESULT,
    ER_NO_PYARROW,
    ER_NO_PYARROW_SNOWSQL,
    ER_NOT_POSITIVE_SIZE,
    ER_UNSUPPORTED_METHOD,
)
from .errors import DatabaseError, Error, InterfaceError, NotSupportedError, ProgrammingError
from .file_transfer_agent import SnowflakeFileTransferAgent
from .json_result import DictJsonResult, JsonResult
from .sqlstate import SQLSTATE_FEATURE_NOT_SUPPORTED
from .telemetry import TelemetryData, TelemetryField
from .time_util import get_time_millis

MYPY = False
if MYPY:  # from typing import TYPE_CHECKING once 3.5 is deprecated
    from .connection import SnowflakeConnection

logger = getLogger(__name__)

try:
    import pyarrow
except ImportError:
    logger.debug("Failed to import pyarrow. Cannot use pandas fetch API")
    pyarrow = None

try:
    from .arrow_result import ArrowResult
    CAN_USE_ARROW_RESULT = True
except ImportError as e:
    logger.debug("Failed to import ArrowResult. No Apache Arrow result set format can be used. ImportError: %s", e)
    CAN_USE_ARROW_RESULT = False

STATEMENT_TYPE_ID_DML = 0x3000
STATEMENT_TYPE_ID_INSERT = STATEMENT_TYPE_ID_DML + 0x100
STATEMENT_TYPE_ID_UPDATE = STATEMENT_TYPE_ID_DML + 0x200
STATEMENT_TYPE_ID_DELETE = STATEMENT_TYPE_ID_DML + 0x300
STATEMENT_TYPE_ID_MERGE = STATEMENT_TYPE_ID_DML + 0x400
STATEMENT_TYPE_ID_MULTI_TABLE_INSERT = STATEMENT_TYPE_ID_DML + 0x500

STATEMENT_TYPE_ID_DML_SET = frozenset(
    [STATEMENT_TYPE_ID_DML, STATEMENT_TYPE_ID_INSERT,
     STATEMENT_TYPE_ID_UPDATE,
     STATEMENT_TYPE_ID_DELETE, STATEMENT_TYPE_ID_MERGE,
     STATEMENT_TYPE_ID_MULTI_TABLE_INSERT])

DESC_TABLE_RE = re.compile(r'desc(?:ribe)?\s+([\w_]+)\s*;?\s*$',
                           flags=re.IGNORECASE)

LOG_MAX_QUERY_LENGTH = 80


def exit_handler(*_):
    """Handler for signal. When called, it will raise SystemExit with exit code FORCE_EXIT."""
    print("\nForce exit")
    logger.info("Force exit")
    sys.exit(1)


class SnowflakeCursor(object):
    """Implementation of Cursor object that is returned from Connection.cursor() method.

    Attributes:
        description: tuple of name, type_code, display_size, internal_size, precisio, scale, null_ok
        rowcount: The number of records updated or selected. If not clear, -1 is returned.
        rownumber: The current 0-based index of the cursor in the result set or None if the index cannot be
            determined.
        sfqid: Snowflake query id in UUID form. Include this in the problem report to the customer support.
        sqlstate: Snowflake SQL State code.
        timestamp_output_format: Snowflake timestamp_output_format for timestamps.
        timestamp_ltz_output_format: Snowflake output format for LTZ timestamps.
        timestamp_tz_output_format: Snowflake output format for TZ timestamps.
        timestamp_ntz_output_format: Snowflake output format for NTZ timestamps.
        date_output_format: Snowflake output format for dates.
        time_output_format: Snowflake output format for times.
        timezone: Snowflake timezone.
        binary_output_format: Snowflake output format for binary fields.
        arrayszize: The default number of rows fetched by fetchmany.
        connection: The connection object by which the cursor was created.
        errorhandle: The class that handles error handling.
        is_file_transfer: Whether, or not the current command is a put, or get.

    TODO:
        Most of these attributes have no reason to be properties, we could just store them in public variables.
        Calling a function is expensive in Python and most of these getters are unnecessary.
    """
    PUT_SQL_RE = re.compile(r'^(?:/\*.*\*/\s*)*put\s+', flags=re.IGNORECASE)
    GET_SQL_RE = re.compile(r'^(?:/\*.*\*/\s*)*get\s+', flags=re.IGNORECASE)
    INSERT_SQL_RE = re.compile(r'^insert\s+into', flags=re.IGNORECASE)
    COMMENT_SQL_RE = re.compile(r"/\*.*\*/")
    INSERT_SQL_VALUES_RE = re.compile(r'.*VALUES\s*(\(.*\)).*',
                                      re.IGNORECASE | re.MULTILINE | re.DOTALL)
    ALTER_SESSION_RE = re.compile(
        r'alter\s+session\s+set\s+(.*)=\'?([^\']+)\'?\s*;',
        flags=re.IGNORECASE | re.MULTILINE | re.DOTALL)

    def __init__(self,
                 connection: 'SnowflakeConnection',
                 use_dict_result: bool = False,
                 json_result_class: object = JsonResult):
        """Inits a SnowflakeCursor with a connection.

        Args:
            connection: The connection that created this cursor.
            use_dict_result: Decides whether to use dict result or not. This variable only applied to
                arrow result. When result in json, json_result_class will be honored.
            json_result_class: The class that used in json result.
        """
        self._connection = connection

        self._errorhandler = Error.default_errorhandler
        self.messages = []
        self._timebomb = None  # must be here for abort_exit method
        self._description = None
        self._column_idx_to_name = None
        self._sfqid = None
        self._sqlstate = None
        self._total_rowcount = -1
        self._sequence_counter = -1
        self._request_id = None
        self._is_file_transfer = False

        self._timestamp_output_format = None
        self._timestamp_ltz_output_format = None
        self._timestamp_ntz_output_format = None
        self._timestamp_tz_output_format = None
        self._date_output_format = None
        self._time_output_format = None
        self._timezone = None
        self._binary_output_format = None
        self._result = None
        self._use_dict_result = use_dict_result
        self._json_result_class = json_result_class

        self._arraysize = 1  # PEP-0249: defaults to 1

        self._lock_canceling = Lock()

        self._first_chunk_time = None

        self._log_max_query_length = connection.log_max_query_length

        self.reset()

    def __del__(self):
        try:
            self.close()
        except BASE_EXCEPTION_CLASS as e:
            if logger.getEffectiveLevel() <= logging.INFO:
                logger.info(e)

    @property
    def description(self):
        return self._description

    @property
    def rowcount(self):
        return self._total_rowcount if self._total_rowcount >= 0 else None

    @property
    def rownumber(self):
        return self._result.total_row_index if self._result.total_row_index >= 0 else None

    @property
    def sfqid(self):
        return self._sfqid

    @property
    def sqlstate(self):
        return self._sqlstate

    @property
    def timestamp_output_format(self):
        return self._timestamp_output_format

    @property
    def timestamp_ltz_output_format(self):
        return self._timestamp_ltz_output_format if \
            self._timestamp_ltz_output_format else \
            self._timestamp_output_format

    @property
    def timestamp_tz_output_format(self):
        return self._timestamp_tz_output_format if \
            self._timestamp_tz_output_format else \
            self._timestamp_output_format

    @property
    def timestamp_ntz_output_format(self):
        return self._timestamp_ntz_output_format if \
            self._timestamp_ntz_output_format else \
            self._timestamp_output_format

    @property
    def date_output_format(self):
        return self._date_output_format

    @property
    def time_output_format(self):
        return self._time_output_format

    @property
    def timezone(self):
        return self._timezone

    @property
    def binary_output_format(self):
        return self._binary_output_format

    @property
    def arraysize(self):
        return self._arraysize

    @arraysize.setter
    def arraysize(self, value):
        self._arraysize = int(value)

    @property
    def connection(self):
        return self._connection

    @property
    def errorhandler(self):
        return self._errorhandler

    @errorhandler.setter
    def errorhandler(self, value):
        logger.debug('setting errorhandler: %s', value)
        if value is None:
            raise ProgrammingError('Invalid errorhandler is specified')
        self._errorhandler = value

    @property
    def is_file_transfer(self):
        """Whether the command is PUT or GET."""
        return hasattr(self, '_is_file_transfer') and self._is_file_transfer

    def callproc(self, procname, args=()):
        """Not supported."""
        Error.errorhandler_wrapper(
            self.connection, self,
            NotSupportedError,
            {
                'msg': "callproc is not supported.",
                'errno': ER_UNSUPPORTED_METHOD,
                'sqlstate': SQLSTATE_FEATURE_NOT_SUPPORTED})

    def close(self):
        """Closes the cursor object."""
        try:
            if self.is_closed():
                return False

            with self._lock_canceling:
                self.reset()
                self._connection = None
                del self.messages[:]
                return True
        except Exception:
            pass

    def is_closed(self):
        return self._connection is None or self._connection.is_closed()

    def _execute_helper(
            self, query, timeout=0, statement_params=None,
            binding_params=None,
            is_internal=False, _no_results=False, _is_put_get=None):
        del self.messages[:]

        if statement_params is not None and not isinstance(
                statement_params, dict):
            Error.errorhandler_wrapper(
                self.connection, self,
                ProgrammingError,
                {
                    'msg': "The data type of statement params is invalid. "
                            "It must be dict.",
                    'errno': ER_INVALID_VALUE,
                })

        # check if current installation include arrow extension or not,
        # if not, we set statement level query result format to be JSON
        if not CAN_USE_ARROW_RESULT:
            logger.debug("Cannot use arrow result format, fallback to json format")
            if statement_params is None:
                statement_params = {PARAMETER_PYTHON_CONNECTOR_QUERY_RESULT_FORMAT: 'JSON'}
            else:
                result_format_val = statement_params.get(PARAMETER_PYTHON_CONNECTOR_QUERY_RESULT_FORMAT)
                if str(result_format_val).upper() == 'ARROW':
                    self.check_can_use_arrow_resultset()
                elif result_format_val is None:
                    statement_params[PARAMETER_PYTHON_CONNECTOR_QUERY_RESULT_FORMAT] = 'JSON'

        self._sequence_counter = self._connection._next_sequence_counter()
        self._request_id = uuid.uuid4()

        if logger.getEffectiveLevel() <= logging.DEBUG:
            logger.debug(
                'running query [%s]', self._format_query_for_log(query))
        if _is_put_get is not None:
            # if told the query is PUT or GET, use the information
            self._is_file_transfer = _is_put_get
        else:
            # or detect it.
            self._is_file_transfer = self.PUT_SQL_RE.match(
                query) or self.GET_SQL_RE.match(query)
        logger.debug('is_file_transfer: %s',
                     self._is_file_transfer is not None)

        real_timeout = timeout if timeout and timeout > 0 \
            else self._connection.network_timeout

        if real_timeout is not None:
            self._timebomb = Timer(
                real_timeout, self.__cancel_query, [query])
            self._timebomb.start()
            logger.debug('started timebomb in %ss', real_timeout)
        else:
            self._timebomb = None

        original_sigint = signal.getsignal(signal.SIGINT)

        def interrupt_handler(*_):
            try:
                signal.signal(signal.SIGINT, exit_handler)
            except (ValueError, TypeError):
                # ignore failures
                pass
            try:
                if self._timebomb is not None:
                    self._timebomb.cancel()
                    logger.debug('cancelled timebomb in finally')
                    self._timebomb = None
                self.__cancel_query(query)
            finally:
                if original_sigint:
                    try:
                        signal.signal(signal.SIGINT, original_sigint)
                    except (ValueError, TypeError):
                        # ignore failures
                        pass
            raise KeyboardInterrupt

        try:
            if not original_sigint == exit_handler:
                signal.signal(signal.SIGINT, interrupt_handler)
        except ValueError:
            logger.debug(
                'Failed to set SIGINT handler. '
                'Not in main thread. Ignored...')
        ret = {'data': {}}
        try:
            ret = self._connection.cmd_query(
                query,
                self._sequence_counter,
                self._request_id,
                binding_params=binding_params,
                is_file_transfer=self._is_file_transfer,
                statement_params=statement_params,
                is_internal=is_internal,
                _no_results=_no_results)
        finally:
            try:
                if original_sigint:
                    signal.signal(signal.SIGINT, original_sigint)
            except (ValueError, TypeError):
                logger.debug(
                    'Failed to reset SIGINT handler. Not in main '
                    'thread. Ignored...')
            except Exception:
                self.connection.incident.report_incident()
                raise
            if self._timebomb is not None:
                self._timebomb.cancel()
                logger.debug('cancelled timebomb in finally')

        if 'data' in ret and 'parameters' in ret['data']:
            for kv in ret['data']['parameters']:
                if 'TIMESTAMP_OUTPUT_FORMAT' in kv['name']:
                    self._timestamp_output_format = kv['value']
                if 'TIMESTAMP_NTZ_OUTPUT_FORMAT' in kv['name']:
                    self._timestamp_ntz_output_format = kv['value']
                if 'TIMESTAMP_LTZ_OUTPUT_FORMAT' in kv['name']:
                    self._timestamp_ltz_output_format = kv['value']
                if 'TIMESTAMP_TZ_OUTPUT_FORMAT' in kv['name']:
                    self._timestamp_tz_output_format = kv['value']
                if 'DATE_OUTPUT_FORMAT' in kv['name']:
                    self._date_output_format = kv['value']
                if 'TIME_OUTPUT_FORMAT' in kv['name']:
                    self._time_output_format = kv['value']
                if 'TIMEZONE' in kv['name']:
                    self._timezone = kv['value']
                if 'BINARY_OUTPUT_FORMAT' in kv['name']:
                    self._binary_output_format = kv['value']
            self._connection._set_parameters(
                ret, self._connection._session_parameters)

        self._sequence_counter = -1
        return ret

    def execute(self,
                command,
                params=None,
                timeout: Optional[int] = None,
                _do_reset: bool = True,
                _put_callback=None,
                _put_azure_callback=None,
                _put_callback_output_stream=sys.stdout,
                _get_callback=None,
                _get_azure_callback=None,
                _get_callback_output_stream=sys.stdout,
                _show_progress_bar: bool = True,
                _statement_params=None,
                _is_internal: bool = False,
                _no_results: bool = False,
                _use_ijson: bool = False,
                _is_put_get: Optional[bool] = None,
                _raise_put_get_error: bool = True,
                _force_put_overwrite: bool = False):
        """Executes a command/query.

        Args:
            command: The SQL command to be executed.
            params: Parameters to be bound into the SQL statement.
            timeout: Number of seconds after which to abort the query.
            _do_reset: Whether or not the result set needs to be reset before executing query.
            _put_callback: Function to which GET command should call back to.
            _put_azure_callback: Function to which an Azure GET command should call back to.
            _put_callback_output_stream: The output stream a PUT command's callback should report on.
            _get_callback: Function to which GET command should call back to.
            _get_azure_callback: Function to which an Azure GET command should call back to.
            _get_callback_output_stream: The output stream a GET command's callback should report on.
            _show_progress_bar: Whether or not to show progress bar.
            _statement_params: Extra information that should be sent to Snowflake with query.
            _no_results: This flag tells the back-end to not return the result, just fire the query and return the
                query id of the running query.
            _use_ijson: This flag doesn't do anything as ijson support has ended.
            _is_put_get: Force decision of this SQL query being a PUT, or GET command. This is detected otherwise.
            _raise_put_get_error: Whether to raise PUT and GET errors.
            _force_put_overwrite: If the SQL query is a PUT, then this flag can force overwriting of an already
                existing file on stage.

        Returns:
            A result class with the results in it. This can either be json, or an arrow result class.
        """
        logger.debug('executing SQL/command')
        if self.is_closed():
            Error.errorhandler_wrapper(
                self.connection, self,
                DatabaseError,
                {'msg': "Cursor is closed in execute.",
                 'errno': ER_CURSOR_IS_CLOSED})

        if _do_reset:
            self.reset()
        command = command.strip(' \t\n\r') if command else None
        if not command:
            logger.warning('execute: no query is given to execute')
            return

        try:
            if self._connection.is_pyformat:
                # pyformat/format paramstyle
                # client side binding
                processed_params = self._connection._process_params(params, self)
                if logger.getEffectiveLevel() <= logging.DEBUG:
                    logger.debug('binding: [%s] with input=[%s], processed=[%s]',
                                 self._format_query_for_log(command),
                                 params, processed_params)
                if len(processed_params) > 0:
                    query = command % processed_params
                else:
                    query = command
                processed_params = None  # reset to None
            else:
                # qmark and numeric paramstyle
                # server side binding
                query = command
                processed_params = self._connection._process_params_qmarks(
                    params, self)
        # Skip reporting Key, Value and Type errors
        except (KeyError, ValueError, TypeError):
            raise
        except Exception:
            self.connection.incident.report_incident()
            raise

        m = DESC_TABLE_RE.match(query)
        if m:
            query1 = 'describe table {}'.format(m.group(1))
            if logger.getEffectiveLevel() <= logging.WARNING:
                logger.info(
                    'query was rewritten: org=%s, new=%s',
                    ' '.join(line.strip() for line in query.split('\n')),
                    query1
                )
            query = query1

        if logger.getEffectiveLevel() <= logging.INFO:
            logger.info(
                'query: [%s]', self._format_query_for_log(query))
        ret = self._execute_helper(
            query,
            timeout=timeout,
            binding_params=processed_params,
            statement_params=_statement_params,
            is_internal=_is_internal,
            _no_results=_no_results,
            _is_put_get=_is_put_get)
        self._sfqid = ret['data']['queryId'] if 'data' in ret and 'queryId' in ret['data'] else None
        self._sqlstate = ret['data']['sqlState'] if 'data' in ret and 'sqlState' in ret['data'] else None
        self._first_chunk_time = get_time_millis()

        # if server gives a send time, log the time it took to arrive
        if 'data' in ret and 'sendResultTime' in ret['data']:
            time_consume_first_result = self._first_chunk_time - ret['data'][
                'sendResultTime']
            self._log_telemetry_job_data(
                TelemetryField.TIME_CONSUME_FIRST_RESULT,
                time_consume_first_result)
        logger.debug('sfqid: %s', self.sfqid)

        logger.info('query execution done')
        if ret['success']:
            logger.debug('SUCCESS')
            data = ret['data']

            # logger.debug(ret)
            logger.debug("PUT OR GET: %s", self.is_file_transfer)
            if self.is_file_transfer:
                sf_file_transfer_agent = SnowflakeFileTransferAgent(
                    self, query, ret,
                    put_callback=_put_callback,
                    put_azure_callback=_put_azure_callback,
                    put_callback_output_stream=_put_callback_output_stream,
                    get_callback=_get_callback,
                    get_azure_callback=_get_azure_callback,
                    get_callback_output_stream=_get_callback_output_stream,
                    show_progress_bar=_show_progress_bar,
                    raise_put_get_error=_raise_put_get_error,
                    force_put_overwrite=_force_put_overwrite or data.get('overwrite', False))
                sf_file_transfer_agent.execute()
                data = sf_file_transfer_agent.result()
                self._total_rowcount = len(data['rowset']) if \
                    'rowset' in data else -1
            m = self.ALTER_SESSION_RE.match(query)
            if m:
                # session parameters
                param = m.group(1).upper()
                value = m.group(2)
                self._connection.converter.set_parameter(param, value)

            if _no_results:
                self._total_rowcount = ret['data'][
                    'total'] if 'data' in ret and 'total' in ret[
                    'data'] else -1
                return data
            self._init_result_and_meta(data)
        else:
            self._total_rowcount = ret['data'][
                'total'] if 'data' in ret and 'total' in ret['data'] else -1
            logger.debug(ret)
            err = ret['message']
            code = ret.get('code', -1)
            if 'data' in ret:
                err += ret['data'].get('errorMessage', '')
            errvalue = {
                'msg': err,
                'errno': int(code),
                'sqlstate': self._sqlstate,
                'sfqid': self._sfqid
            }
            Error.errorhandler_wrapper(self.connection, self,
                                       ProgrammingError,
                                       errvalue)
        return self

    def _format_query_for_log(self, query):
        return self._connection._format_query_for_log(query)

    def _is_dml(self, data):
        return 'statementTypeId' in data \
               and int(data['statementTypeId']) in \
               STATEMENT_TYPE_ID_DML_SET

    def _init_result_and_meta(self, data):
        is_dml = self._is_dml(data)
        self._query_result_format = data.get('queryResultFormat', 'json')
        logger.debug("Query result format: %s", self._query_result_format)

        if self._total_rowcount == -1 and not is_dml and data.get('total') \
                is not None:
            self._total_rowcount = data['total']

        self._description = []

        for column in data['rowtype']:
            type_value = FIELD_NAME_TO_ID[column['type'].upper()]
            self._description.append((column['name'],
                                      type_value,
                                      None,
                                      column['length'],
                                      column['precision'],
                                      column['scale'],
                                      column['nullable']))

        if self._query_result_format == 'arrow':
            self.check_can_use_arrow_resultset()
            self._result = ArrowResult(data, self, use_dict_result=self._use_dict_result)
        else:
            self._result = self._json_result_class(data, self)

        if is_dml:
            updated_rows = 0
            for idx, desc in enumerate(self._description):
                if desc[0] in (
                        'number of rows updated',
                        'number of multi-joined rows updated',
                        'number of rows deleted') or \
                        desc[0].startswith('number of rows inserted'):
                    updated_rows += int(data['rowset'][0][idx])
            if self._total_rowcount == -1:
                self._total_rowcount = updated_rows
            else:
                self._total_rowcount += updated_rows

    def check_can_use_arrow_resultset(self):
        global CAN_USE_ARROW_RESULT

        if not CAN_USE_ARROW_RESULT:
            if self._connection.application == 'SnowSQL':
                msg = (
                    "Currently SnowSQL doesn't support the result set in Apache Arrow format."
                )
                errno = ER_NO_PYARROW_SNOWSQL
            else:
                msg = (
                    "The result set in Apache Arrow format is not supported for the platform."
                )
                errno = ER_NO_ARROW_RESULT

            Error.errorhandler_wrapper(
                self.connection, self,
                ProgrammingError,
                {
                    'msg': msg,
                    'errno': errno,
                }
            )

    def check_can_use_pandas(self):
        global pyarrow

        if pyarrow is None:
            msg = ("Optional dependency: 'pyarrow' is not installed, please see the following link for install "
                   "instructions: https://docs.snowflake.com/en/user-guide/python-connector-pandas.html#installation")
            errno = ER_NO_PYARROW

            Error.errorhandler_wrapper(
                self.connection, self,
                ProgrammingError,
                {
                    'msg': msg,
                    'errno': errno,
                }
            )

    def query_result(self, qid):
        url = '/queries/{qid}/result'.format(qid=qid)
        ret = self._connection.rest.request(url=url, method='get')
        self._sfqid = ret['data'][
            'queryId'] if 'data' in ret and 'queryId' in ret[
            'data'] else None
        self._sqlstate = ret['data'][
            'sqlState'] if 'data' in ret and 'sqlState' in ret[
            'data'] else None
        logger.debug('sfqid=%s', self._sfqid)

        if ret.get('success'):
            data = ret.get('data')
            self._init_result_and_meta(data)
        else:
            logger.info('failed')
            logger.debug(ret)
            err = ret['message']
            code = ret.get('code', -1)
            if 'data' in ret:
                err += ret['data'].get('errorMessage', '')
            errvalue = {
                'msg': err,
                'errno': int(code),
                'sqlstate': self._sqlstate,
                'sfqid': self._sfqid
            }
            Error.errorhandler_wrapper(self.connection, self,
                                       ProgrammingError,
                                       errvalue)
        return self

    def fetch_pandas_batches(self, **kwargs):
        """Fetches a single Arrow Table."""
        self.check_can_use_pandas()
        if self._query_result_format != 'arrow':  # TODO: or pandas isn't imported
            raise NotSupportedError
        for df in self._result._fetch_pandas_batches(**kwargs):
            yield df

    def fetch_pandas_all(self, **kwargs):
        """Fetch Pandas dataframes in batches, where 'batch' refers to Snowflake Chunk."""
        self.check_can_use_pandas()
        if self._query_result_format != 'arrow':
            raise NotSupportedError
        return self._result._fetch_pandas_all(**kwargs)

    def abort_query(self, qid):
        url = '/queries/{qid}/abort-request'.format(qid=qid)
        ret = self._connection.rest.request(url=url, method='post')
        return ret.get('success')

    def executemany(self, command, seqparams):
        """Executes a command/query with the given set of parameters sequentially."""
        logger.debug('executing many SQLs/commands')
        command = command.strip(' \t\n\r') if command else None

        if len(seqparams) == 0:
            errorvalue = {
                'msg': "No parameters are specified for the command: "
                        "{}".format(command),
                'errno': ER_INVALID_VALUE,
            }
            Error.errorhandler_wrapper(
                self.connection, self, InterfaceError, errorvalue
            )
            return self

        if self.INSERT_SQL_RE.match(command):
            if self._connection.is_pyformat:
                logger.debug('rewriting INSERT query')
                command_wo_comments = re.sub(self.COMMENT_SQL_RE, '', command)
                m = self.INSERT_SQL_VALUES_RE.match(command_wo_comments)
                if not m:
                    errorvalue = {
                        'msg': "Failed to rewrite multi-row insert",
                        'errno': ER_FAILED_TO_REWRITE_MULTI_ROW_INSERT
                    }
                    Error.errorhandler_wrapper(
                        self.connection, self, InterfaceError, errorvalue
                    )

                fmt = m.group(1)
                values = []
                for param in seqparams:
                    logger.debug('parameter: %s', param)
                    values.append(fmt % self._connection._process_params(
                        param, self))
                command = command.replace(fmt, ','.join(values), 1)
                self.execute(command)
                return self
            else:
                logger.debug('bulk insert')
                num_params = len(seqparams[0])
                pivot_param = []
                for _ in range(num_params):
                    pivot_param.append([])
                for row in seqparams:
                    if len(row) != num_params:
                        errorvalue = {
                            'msg':
                                "Bulk data size don't match. expected: {}, "
                                "got: {}, command: {}".format(
                                    num_params, len(row), command),
                            'errno': ER_INVALID_VALUE,
                        }
                        Error.errorhandler_wrapper(
                            self.connection, self, InterfaceError, errorvalue
                        )
                        return self
                    for idx, value in enumerate(row):
                        pivot_param[idx].append(value)
                self.execute(command, params=pivot_param)
                return self

        self.reset()
        for param in seqparams:
            self.execute(command, param, _do_reset=False)
        return self

    def fetchone(self):
        """Fetches one row."""
        try:
            return next(self._result)
        except StopIteration:
            return None

    def fetchmany(self, size=None):
        """Fetches the number of specified rows."""
        if size is None:
            size = self.arraysize

        if size < 0:
            errorvalue = {
                'msg': ("The number of rows is not zero or "
                         "positive number: {0}").format(
                    size),
                'errno': ER_NOT_POSITIVE_SIZE}
            Error.errorhandler_wrapper(
                self.connection, self, ProgrammingError, errorvalue)
        ret = []
        while size > 0:
            row = self.fetchone()
            if row is None:
                break
            ret.append(row)
            if size is not None:
                size -= 1

        return ret

    def fetchall(self):
        """Fetches all of the results."""
        ret = []
        while True:
            row = self.fetchone()
            if row is None:
                break
            ret.append(row)
        return ret

    def nextset(self):
        """Not supported."""
        logger.debug('nop')
        return None

    def setinputsizes(self, _):
        """Not supported."""
        logger.debug('nop')

    def setoutputsize(self, _, column=None):
        """Not supported."""
        del column
        logger.debug('nop')

    def scroll(self, value, mode='relative'):
        Error.errorhandler_wrapper(
            self.connection, self,
            NotSupportedError,
            {
                'msg': "scroll is not supported.",
                'errno': ER_UNSUPPORTED_METHOD,
                'sqlstate': SQLSTATE_FEATURE_NOT_SUPPORTED})

    def reset(self):
        """Resets the result set."""
        self._total_rowcount = -1  # reset the rowcount
        if self._result is not None:
            self._result._reset()

    def __iter__(self):
        """Iteration over the result set."""
        return iter(self._result)

    def __cancel_query(self, query):
        if self._sequence_counter >= 0 and not self.is_closed():
            logger.debug('canceled. %s, request_id: %s',
                         query, self._request_id)
            with self._lock_canceling:
                self._connection._cancel_query(query, self._request_id)

    def _log_telemetry_job_data(self, telemetry_field, value):
        """Builds an instance of TelemetryData with the given field and logs it."""
        obj = {
            'type': telemetry_field,
            'query_id': self._sfqid,
            'value': int(value)
        }
        ts = get_time_millis()
        try:
            self._connection._log_telemetry(TelemetryData(obj, ts))
        except AttributeError:
            logger.warning(
                "Cursor failed to log to telemetry. Connection object may be None.",
                exc_info=True)

    def __enter__(self):
        """Context manager."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Context manager with commit or rollback."""
        self.close()


class DictCursor(SnowflakeCursor):
    """Cursor returning results in a dictionary."""

    def __init__(self, connection):
        SnowflakeCursor.__init__(self, connection, use_dict_result=True, json_result_class=DictJsonResult)