def METHOD_NAME(
    self,
    operation: str,
    parameters: Optional[Dict[str, Any]] = None,
    work_group: Optional[str] = None,
    s3_staging_dir: Optional[str] = None,
    cache_size: int = 0,
    cache_expiration_time: int = 0,
    result_reuse_enable: Optional[bool] = None,
    result_reuse_minutes: Optional[int] = None,
    **kwargs,
) -> ArrowCursor:
    self._reset_state()
    if self._unload:
        s3_staging_dir = s3_staging_dir if s3_staging_dir else self._s3_staging_dir
        assert s3_staging_dir, "If the unload option is used, s3_staging_dir is required."
        operation, unload_location = self._formatter.wrap_unload(
            operation,
            s3_staging_dir=s3_staging_dir,
            format_=AthenaFileFormat.FILE_FORMAT_PARQUET,
            compression=AthenaCompression.COMPRESSION_SNAPPY,
        )
    else:
        unload_location = None
    self.query_id = self._execute(
        operation,
        parameters=parameters,
        work_group=work_group,
        s3_staging_dir=s3_staging_dir,
        cache_size=cache_size,
        cache_expiration_time=cache_expiration_time,
        result_reuse_enable=result_reuse_enable,
        result_reuse_minutes=result_reuse_minutes,
    )
    query_execution = self._poll(self.query_id)
    if query_execution.state == AthenaQueryExecution.STATE_SUCCEEDED:
        self.result_set = AthenaArrowResultSet(
            connection=self._connection,
            converter=self._converter,
            query_execution=query_execution,
            arraysize=self.arraysize,
            retry_config=self._retry_config,
            unload=self._unload,
            unload_location=unload_location,
            **kwargs,
        )
    else:
        raise OperationalError(query_execution.state_change_reason)
    return self