o FDiR@sddlZddlZddlZddlmZddlmZddlmZddlm Z m Z m Z m Z m Z ddlmZddlmZmZmZmZddlmZdd lmZmZmZmZdd lmZdd lmZm Z dd l!m"Z"m#Z#m$Z$dd l%m&Z&m'Z'ddl(m)Z)ddl*m+Z+m,Z,m-Z-ddl.m/Z/m0Z0m1Z1m2Z2ddl3m4Z4ddl5m6Z6m7Z7ddl8m9Z9m:Z:ddl;mZ>m?Z?e@eAZBGdddZCGdddZDGdddZEdS)N)AsyncGenerator)time) GeneratorType)AnyIterableMappingOptionalUnion)urlparse)PartiallyConsumedQueryErrorServerExceptionUnexpectedPacketFromServerErrorUnknownPacketFromServerError) constants) BaseBlockBlockStreamProfileInfoColumnOrientedBlockRowOrientedBlock)get_compressor_cls)ContextExecuteContext) ClientInfo QueryKind ServerInfo)ClickhouseSchemeCompressionAlgorithm)Progress) ClientPacket Compression ServerPacket)IterQueryResultProgressQueryResult QueryInfo QueryResult)write_settings) BlockReader BlockWriter)BufferedReaderBufferedWriter) escape_params)chunks column_chunksc@seZdZdZdZdZdZdS)QueryProcessingStagezd?Z)dwdAedBefdCdDZ*dedefdEdFZ+dGdHZ,dIdJZ-dKdLZ.    @   dxdMdNZ/    @   dxdOdPZ0      dydAedQe1e2ee3fdRedSe1e4e2dBe1edTedUefdVdWZ5dXdYZ6dZd[Z7d\d]Z8      dydAedQe1e2ee3fdRedSe1e4e2dBe1edTedUefd^d_Z9dtd`daZ:dwdbdcZ;edhdiZ?d{djdkZ@     d|dAedQe1e2ee3fdRedSe1e4e2dBe1edTef dldmZAdtdndoZBdpdqZC    @  d}drdsZDdS)~ Connection) UnknownFatalCriticalErrorWarningNotice InformationDebugTraceFTNuserpasswordhostportdatabase client_nameconnect_timeoutsend_receive_timeoutsync_request_timeoutcompress_block_size compressionsecureverify alt_hostsc Ks&||_| r tj}ntj}||p|fg|_|r4|dD]}ttjd|}|j |j |j p0|fq||_ d|_ d|_ ||_||_tjd||_||_||_| |_||_t|_| |_| |_i}|durl||d<|durt||d<|dur|||d<||_| durtj} | durtj|_ d|_!d|_"n tj#|_ t$| |_!| |_"d|_%d|_&d|_'d|_(t)|_*d|_+d|_,d|_-d|_.d|_/|0d i1|_2t3|j20d tj4|j20d d|j20d tj5|j20d d|j20dd|j20dd|j20dd|j20ddd|_6d|_7d|_8|j2|j*_2|j6|j*_6dS)N,z:// ssl_versionca_certsciphersTFsettingsinsert_block_sizestrings_as_bytesstrings_encoding use_numpyopentelemetry_traceparentopentelemetry_tracestate quota_keyinput_format_null_as_default)rarbrcrdrerfrhri)9 stack_trackrDEFAULT_SECURE_PORT DEFAULT_PORThostssplitr r clickhouseappendhostnamerPrQrOrMrN DBMS_NAMErRrSrTrUsettings_is_importantasyncioLock_lock secure_socketrY ssl_optionsrlz4rDISABLEDrWcompressor_clsrVENABLEDr connectedreaderwriter server_inforcontext block_reader block_writerblock_reader_rawis_query_executingclient_trace_contextpopcopyr`intDEFAULT_INSERT_BLOCK_SIZESTRINGS_ENCODINGclient_settings last_queryavailable_client_settings)rArMrNrOrPrQrRrSrTrUrVrWrXrYr]r^r_rZrjrskwargs default_porturlrxr6r6r7rBKs        zConnection.__init__cCsB|jrddlm}||j|j|jSddlm}||j|j|jS)Nr)CompressedBlockReader)r%)rWasynch.proto.streams.compressedrr~rrasynch.proto.streams.blockr%)rArr%r6r6r7get_block_readers  zConnection.get_block_readercCsT|jr|t}ddlm}||j|j|j||jSddl m }||j|j|jS)Nr-)CompressedBlockWriter)r&) rWr{r(streams.compressedrr~rrrV streams.blockr&)rA compressorrr&r6r6r7get_block_writers   zConnection.get_block_writercs|jtjIdH|j|jIdH|jtjIdH|jtjIdH|jtj IdH|j|j IdH|j|j IdH|j|j IdH|j IdHdSr9)r write_varintrHELLO write_strrRrCLIENT_VERSION_MAJORCLIENT_VERSION_MINORCLIENT_REVISIONrQrMrNflushr@r6r6r7 send_helloszConnection.send_hellocCs|jd|jS)N:)rOrPr@r6r6r7 get_serverszConnection.get_servercCst|}d|||S)Nz6Unexpected packet from server {} (expected {}, got {}))rto_strformatr)rAexpected packet_typer6r6r7unexpected_packet_messages  z$Connection.unexpected_packet_messagec s(|jIdH}|tjkrw|jIdH}|jIdH}|jIdH}|jIdH}d}|tjkr=|jIdH}d}|tjkrL|jIdH}|}|tjkr[|jIdH}t ||||||||_ |j |j _ t d|||||dS|tjkr|IdH|IdH|d|} t| )Nrgz5Connected to %s server version %s.%s.%s, revision: %szHello or Exception)r~ read_varintrrread_strr&DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE*DBMS_MIN_REVISION_WITH_SERVER_DISPLAY_NAME$DBMS_MIN_REVISION_WITH_VERSION_PATCHrrrloggerdebug EXCEPTIONread_exception disconnectrr ) rAr server_nameserver_version_majorserver_version_minorserver_revisionserver_timezoneserver_display_nameserver_version_patchmessager6r6r7 receive_hellosN       zConnection.receive_hellocCs|jsdSttj}|jd}|r|j|O_|jd}|r(||n|tj j |jd}|r<| ||j rFtj j|_|Sd|_tj j|_|S)Nr]r^r_F)rwssl SSLContextPROTOCOL_TLS_CLIENTrxgetoptionsload_verify_locationsload_default_certsPurpose SERVER_AUTH set_ciphersrY VerifyMode CERT_REQUIRED verify_modecheck_hostname CERT_NONE)rAssl_ctxr]r^r_r6r6r7_get_ssl_contexts&        zConnection._get_ssl_contextreturnc s`z\|jjrtd|j|IdHWdS|jtjIdH|j IdH|j IdH}|t j krL| IdH|j IdH}|t j ks8|t jkr[|d|}t|WdStymtd|YdSty}ztjd|dWYd}~dSd}~wtttfy}zt|trdt|vr|tjd |dWYd}~dSd}~ww) Nz %s at EOFFPongTzThe connection %s is not openzPing package smaller than expected or empty. There may be connection or network problems - we believe that the connection is incorrect.exc_infozTCPTransport closed=True Socket closed)r~at_eofrrrrrrPINGrrrPROGRESSreceive_progressPONGrr AttributeError IndexErrorConnectionErrorOSError RuntimeError isinstancestr)rArmsger6r6r7ping2sD      zConnection.pingcs>|jj}|tjkr|jIdH|r|jn|jIdHSr9) rrevisionr'DBMS_MIN_REVISION_WITH_TEMPORARY_TABLESr~rrrread)rArawrr6r6r7 receive_dataTs  zConnection.receive_datacs|IdHSr9)rr@r6r6r7receive_exceptionZszConnection.receive_exceptionc s|jIdH}|jIdH}|jIdH}|jIdH}t|jIdH}d}|r5||d7}|dkr?||d7}||7}|jrL|d|7}d}|rW|IdH}t|||dS)Nrgz. z DB::Exceptionz. Stack trace: )nested)r~ read_int32rbool read_uint8rjrr ) rAadditional_messagecodenamer stack_trace has_nested new_messagerr6r6r7r]s$   zConnection.read_exceptioncst|j}|IdH|Sr9)rr~r)rAr>r6r6r7receive_profile_infous zConnection.receive_profile_inforcs(t|}fddt|DIdHS)Ncsg|] }jIdHqSr9)r~r).0_ir@r6r7 |sz:Connection.receive_multistring_message..)rstrings_in_messagerange)rArnumr6r@r7receive_multistring_messagezs z&Connection.receive_multistring_messager;c Csdd|jD}|D]?}tt||}d|dkr!dkr+nn|j|d}n|d}|dp7|d}td |d ||d ||d |d q dS)NcSsg|]}|dqS)rr6)rxr6r6r7rsz(Connection.log_block..r-priorityr thread_id thread_numberz[ %s ] [ %s ] {%s} <%s> %s: %s host_namequery_idsourcetext)columns_with_typesget_rowsdictziplog_prioritiesrrinfo)rAr; column_namesrowrrr6r6r7 log_block~s" zConnection.log_blockcs|IdH}|jtjkr|j|jtjkr |j|jdS|jtj kr(dS|jtj kr0|S|jtj kr8|S|jtj kr@|S|jtj krO|j|jdSdS)NTF)_receive_packetr:rrr<rrstore_progressr= END_OF_STREAMDATATOTALSEXTREMES PROFILE_INFO store_profiler>rApacketr6r6r7receive_packets&       zConnection.receive_packetcst}|jIdH|_}|tjkr|IdH|_|S|tjkr-| IdH|_ |S|jtj kr=| IdH|_ |S|jtjkrM|IdH|_|S|tjkr\|IdH|_|S|tjkrk|IdH|_|S|tjkr|jddIdH}|||S|tjkrd|_|S|tjkr||IdH|_|S|tjkr|IdH|_|S|tjkr|IdH|_|S|tjkr|jddIdH|_|S|IdHtd|d|)NT)rFzUnknown packet z from server ) r8r~rr:rrrr;rrr<rrr=rrr>rrLOGr rr TABLE_COLUMNSrr? PART_UUIDSREAD_TASK_REQUESTPROFILE_EVENTSrrr)rArrr;r6r6r7r sX $ !           zConnection._receive_packetcCs, |IdH}|s dS|durq|VqNT)rrr6r6r7packet_generatorszConnection.packet_generatorcs@|}|rt|j|||dSt|j|||d}|IdHS)Nwith_column_typescolumnar)rr!r~r# get_result)rArr=r  generatorresultr6r6r7receive_results  zConnection.receive_resultcs$t|j}||jjIdH|Sr9)rr~rrr)rAr=r6r6r7rs   zConnection.receive_progresscs*|jtjIdH|jIdHdSr9)rrrCANCELrr@r6r6r7 send_cancelszConnection.send_cancelrgqueryrcs|jtjIdH|j|IdH|jj}|tjkr2t |j |j|j }t j |_||IdH|tjk}t|j|j j||jIdH|tjkrS|jdIdH|jtjIdH|j|jIdH|j|IdHtd||jIdHdS)Nrgz Query: %s)rrrQUERYrrrr"DBMS_MIN_REVISION_WITH_CLIENT_INFOrrRrr INITIAL_QUERY query_kindwrite5DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGSr$r`rs)DBMS_MIN_REVISION_WITH_INTERSERVER_SECRETr,r5rWrrr)rAr'rr client_infosettings_as_stringsr6r6r7 send_querys(   zConnection.send_querycs|||_|_tj|||dIdH\}}t||_t||_| |_ t |j|j|j |_ ||_d|_|IdH|IdHdS)N)rT)rOrPrtopen_connectionrr(rr'r~rrr%rrrrr}rr)rArOrPr~rr6r6r7_init_connections    zConnection._init_connectioncCs:d|_d|_d|_d|_d|_d|_d|_d|_d|_dS)NF) rr~rrrr}rrrr@r6r6r7 reset_state*s zConnection.reset_statec sf|jr1z |jIdHWnty'}z tjd|dWYd}~nd}~ww|d|_dSdS)NrrF)r}rcloserrrr4)rArr6r6r7r7s zConnection.disconnectcs^|jr |IdHtd|j|j|jD]\}}td|||||IdHSdS)Nz"Connecting. Database: %s. User: %szConnecting to %s:%s)r}rrrrQrMrmr3)rArOrPr6r6r7connectAszConnection.connectc st} t|||4IdH>t|tttf} | r(|j||||||dIdH} n|j|||||||dIdH} |j t| | WdIdHS1IdHsRwYdS)a Executes query. Establishes new connection if it wasn't established yet. After query execution connection remains intact for next queries. If connection can't be reused it will be closed and new connection will be created. :param query: query that will be send to server. :param args: substitution parameters for SELECT queries and data for INSERT queries. Data for INSERT can be `list`, `tuple` or :data:`~types.GeneratorType`. Defaults to ``None`` (no parameters or data). :param with_column_types: if specified column names and types will be returned alongside with result. Defaults to ``False``. :param external_tables: external tables to send. Defaults to ``None`` (no external tables). :param query_id: the query identifier. If no query id specified ClickHouse server will generate it. :param settings: dictionary of query settings. Defaults to ``None`` (no additional settings). :param types_check: enables type checking of data for INSERT queries. Causes additional overhead. Defaults to ``False``. :param columnar: if specified the result of the SELECT query will be returned in column-oriented form. It also allows to INSERT data in columnar form. Defaults to ``False`` (row-like form). :return: * number of inserted rows for INSERT queries with data. Returning rows count from INSERT FROM SELECT is not supported. * if `with_column_types=False`: `list` of `tuples` with rows/columns. * if `with_column_types=True`: `tuple` of 2 elements: * The first element is `list` of `tuples` with rows/columns. * The second element information is about columns: names and types. Nexternal_tablesr types_checkr )rr8rr9r ) rrrlisttuplerprocess_insert_queryprocess_ordinary_queryr store_elapsed) rAr'argsrr8rr`r9r  start_time is_insertrvr6r6r7executeIs24   0zConnection.executec sdt|||4IdH|j|||||||dIdHWdIdHS1IdHs+wYdS)a Executes SELECT query with progress information. See, :ref:`execute-with-progress`. :param query: query that will be send to server. :param args: substitution parameters for SELECT queries and data for INSERT queries. Data for INSERT can be `list`, `tuple` or :data:`~types.GeneratorType`. Defaults to ``None`` (no parameters or data). :param with_column_types: if specified column names and types will be returned alongside with result. Defaults to ``False``. :param external_tables: external tables to send. Defaults to ``None`` (no external tables). :param query_id: the query identifier. If no query id specified ClickHouse server will generate it. :param settings: dictionary of query settings. Defaults to ``None`` (no additional settings). :param types_check: enables type checking of data for INSERT queries. Causes additional overhead. Defaults to ``False``. :param columnar: if specified the result will be returned in column-oriented form. Defaults to ``False`` (row-like form). :return: :ref:`progress-query-result` proxy. N)paramsrr8rr9r )r$process_ordinary_query_with_progress) rAr'r?rr8rr`r9r r6r6r7execute_with_progresss% 0z Connection.execute_with_progressrDrr8r9r csN|dur |||}|j||dIdH|j||dIdH|j|d|dS)Nrr9T)rr=r substitute_paramsr1send_external_tablesr$rAr'rDrr8rr9r r6r6r7rEs z/Connection.process_ordinary_query_with_progresscCs`t|pi}|j}|jD] }||vr||||<q||j_|j}||||j_dSr9)rrrrrrr`update)rAr`rkeyquery_settingsr6r6r7make_query_settingss      zConnection.make_query_settingsc sT|j4IdH|jrtd|_WdIdHdS1IdHs#wYdSr)rvrr r@r6r6r7check_query_executions .z Connection.check_query_executioncsV|IdH|js|IdHdS|IdHs)td|IdHdSdS)Nz$Connection was closed, reconnecting.)rQr}r6rrrr@r6r6r7 force_connects zConnection.force_connectcsR|dur |||}|j||dIdH|j||dIdH|j||dIdHS)NrGrHrrIrLr6r6r7r=s  z!Connection.process_ordinary_querycsp|pgD]'}|dstd|ddt|d|d|d}|j||ddIdHq|tIdHdS)N structurez Empty table "rz " structuredatarH) table_name) ValueErrorr send_block)rAtablesr9tabler;r6r6r7rKs zConnection.send_external_tablescsP|jtjIdH|jj}|tjkr|j|IdH|j |IdHdSr9) rrrrrrrrrrr,)rAr;rUrr6r6r7rWs  zConnection.send_blockcCs&t|ts td|jdit|S)Nz'Parameters are expected to be a mappingr6)rrrVrr))r'rDr6r6r7rJ$s zConnection.substitute_paramsc sv|j||dIdH|j||dIdH|IdH}|r9|j||||dIdH}|IdH} | jr7| j|SdS)NrGrH)r9r )r1rKreceive_sample_block send_datar r<) rAquery_without_datarTr8rr9r  sample_blockrBrr6r6r7r<+s  zConnection.process_insert_querycsv |IdH}|jtjkr|jS|jtjkr|j|jtjkr(||jn|jtj kr/n | d|j}t |q)NTzData, Exception or TableColumns) r r:rrr;rr<rr rrr )rArrr6r6r7rZAs     zConnection.receive_sample_blockc szd}|jj}|r tnt}|rtnt}|||dD]} ||j| |d} || IdH|| j7}q||IdH|S)Nrra)rrTr9) rrrrr+r*rrWnum_rows) rAr]rTr9r  inserted_rowsr block_clsslicerchunkr;r6r6r7r[Us   zConnection.send_datacsJ|dur |||}|j||dIdH|j||dIdH|j|dS)NrGrHr)rJr1rKiter_receive_result)rAr'rDrr8rr9r6r6r7iter_process_ordinary_queryis   z&Connection.iter_process_ordinary_querycCs|}t||dS)Nrc)rr )rArgenr6r6r7rdys zConnection.iter_receive_resultcCs2|d}|dr|dd|_dSdS)Nz; zuse )striplower startswithrQ)rAr'r6r6r7track_current_database~s z!Connection.track_current_databasec st|||4IdH;t|tttf}|r.|j|||||ddIdHWdIdHS|j||||||dIdHWdIdHS1IdHsLwYdS)a *New in version 0.0.14.* Executes SELECT query with results streaming. See, :ref:`execute-iter`. :param query: query that will be send to server. :param args: substitution parameters for SELECT queries and data for INSERT queries. Data for INSERT can be `list`, `tuple` or :data:`~types.GeneratorType`. Defaults to ``None`` (no parameters or data). :param with_column_types: if specified column names and types will be returned alongside with result. Defaults to ``False``. :param external_tables: external tables to send. Defaults to ``None`` (no external tables). :param query_id: the query identifier. If no query id specified ClickHouse server will generate it. :param settings: dictionary of query settings. Defaults to ``None`` (no additional settings). :param types_check: enables type checking of data for INSERT queries. Causes additional overhead. Defaults to ``False``. :return: :ref:`iter-query-result` proxy. NFr7)rDrr8rr9)rrr:r;rr<re) rAr'r?rr8rr`r9rAr6r6r7 execute_iters,"   0zConnection.execute_iter)Fr9)FFF)rg)NFNrgNFF)NFNNFF)NNFF)FF)NFNNF)NFNrgNF)Er/r0r1rr DEFAULT_USERDEFAULT_PASSWORD DEFAULT_HOSTrlDEFAULT_DATABASE CLIENT_NAME DBMS_DEFAULT_CONNECT_TIMEOUT_SECDBMS_DEFAULT_TIMEOUT_SEC%DBMS_DEFAULT_SYNC_REQUEST_TIMEOUT_SECDEFAULT_COMPRESS_BLOCK_SIZErrr rrBrrrrrrrrrrrrrrr rr rrr$rr&r1r3r4rr6rCrFrrrrrErPrQrRr=rKrW staticmethodrJr<rZr[rerdrkrlr6r6r6r7rC>s\      p - " ,      S 3            rC)Frtloggingrcollections.abcrrtypesrtypingrrrrr urllib.parser asynch.errorsr r r r asynch.protorasynch.proto.blockrrrrasynch.proto.compressionrasynch.proto.contextrrasynch.proto.csrrrasynch.proto.models.enumsrrasynch.proto.progressrasynch.proto.protocolrrrasynch.proto.resultr r!r"r#asynch.proto.settingsr$rr%r&asynch.proto.streams.bufferedr'r(asynch.proto.utils.escaper)asynch.proto.utils.helpersr*r+ getLoggerr/rr,r8rCr6r6r6r7s6