o ȳg`@sRddlmZddlmZmZmZmZmZer ddlm Z m Z m Z GdddZ dS)) annotations) TYPE_CHECKINGAnyIterableListOptional) DataFrameRow SparkSessionc@seZdZdZ      d9d:ddZe d;ddd Z d;d?d"d#Z d>d$d%Z d@d)d*Z dAd.d/Z dBdCd3d4Zd;d?d5d6ZdBdCd7d8ZdS)DSparkSQLz;SparkSQL is a utility class for interacting with Spark SQL.N spark_sessionOptional[SparkSession]catalog Optional[str]schema ignore_tablesOptional[List[str]]include_tablessample_rows_in_table_infointc Cs"zddlm}Wn tytdw|r|n|j|_|dur)|jj||dur4|jj|t | |_ |rAt |nt |_ |j rX|j |j }|rXt d|d|r^t |nt |_|jru|j|j }|rut d|d|} | rt | n|j |_t|tstd||_dS) aInitialize a SparkSQL object. Args: spark_session: A SparkSession object. If not provided, one will be created. catalog: The catalog to use. If not provided, the default catalog will be used. schema: The schema to use. If not provided, the default schema will be used. ignore_tables: A list of tables to ignore. If not provided, all tables will be used. include_tables: A list of tables to include. If not provided, all tables will be used. sample_rows_in_table_info: The number of rows to include in the table info. Defaults to 3. rr Fpyspark is not installed. Please install it with `pip install pyspark`Nzinclude_tables  not found in databasezignore_tables z,sample_rows_in_table_info must be an integer) pyspark.sqlr ImportErrorbuilder getOrCreate_sparkrsetCurrentCatalogsetCurrentDatabaseset_get_all_table_names _all_tables_include_tables ValueError_ignore_tablesget_usable_table_names_usable_tables isinstancer TypeError_sample_rows_in_table_info) selfr rrrrrr missing_tables usable_tablesr/c/var/www/html/chatdoc2/venv/lib/python3.10/site-packages/langchain_community/utilities/spark_sql.py__init__ sB       zSparkSQL.__init__ database_uristr engine_argsOptional[dict]kwargsrreturncKsHzddlm}Wn tytdw|j|}||fi|S)zzCreating a remote Spark Session via Spark connect. For example: SparkSQL.from_uri("sc://localhost:15002") rrr)rr rrremoter)clsr2r4r6r sparkr/r/r0from_uriKs zSparkSQL.from_uri Iterable[str]cCs|jr|jSt|j|jS)zGet names of tables available.)r$sortedr#r&)r,r/r/r0r'\szSparkSQL.get_usable_table_namescCs(|jdd}ttdd|S)Nz SHOW TABLES tableNamecS|jSN)r>)rowr/r/r0ez/SparkSQL._get_all_table_names..)rsqlselectcollectlistmap)r,rowsr/r/r0r"cszSparkSQL._get_all_table_namestablecCs6|jd|dj}|d}|d|dS)NzSHOW CREATE TABLE rUSING;)rrDrFcreatetab_stmtfind)r,rJ statementusing_clause_indexr/r/r0_get_create_table_stmtgs zSparkSQL._get_create_table_stmt table_namescCs|}|durt||}|rtd|d|}g}|D]"}||}|jr<|d7}|d||d7}|d7}||qd|}|S)Nz table_names rz /* z*/z ) r'r! differencer%rQr+_get_sample_spark_rowsappendjoin)r,rRall_table_namesr-tables table_name table_info final_strr/r/r0get_table_infoos    zSparkSQL.get_table_infocCsd|d|j}|j|}dttdd|jj}z||}ddd|D}Wn t y9d }Ynw|jd |d |d|S) NzSELECT * FROM z LIMIT  cSr?r@)name)fr/r/r0rBrCz1SparkSQL._get_sample_spark_rows..rScSsg|]}d|qS)r^)rW).0rAr/r/r0 sz3SparkSQL._get_sample_spark_rows..z rows from z table: ) r+rrDrWrGrHrfields_get_dataframe_results Exception)r,rJquerydf columns_str sample_rowssample_rows_strr/r/r0rUs   zSparkSQL._get_sample_spark_rowsrAr tuplecCsttt|Sr@)rlrHr3asDictvalues)r,rAr/r/r0_convert_row_as_tupleszSparkSQL._convert_row_as_tuplerhrrGcCstt|j|Sr@)rGrHrorF)r,rhr/r/r0reszSparkSQL._get_dataframe_resultsallcommandfetchcCs,|j|}|dkr|d}t||S)None)rrDlimitr3re)r,rqrrrhr/r/r0runs  z SparkSQL.runc Cs>z||WSty}z d|WYd}~Sd}~ww)afGet information about specified tables. Follows best practices as specified in: Rajkumar et al, 2022 (https://arxiv.org/abs/2204.00498) If `sample_rows_in_table_info`, the specified number of sample rows will be appended to each table description. This can increase performance as demonstrated in the paper. Error: N)r]r%)r,rRer/r/r0get_table_info_no_throws z SparkSQL.get_table_info_no_throwc Cs@z|||WSty}z d|WYd}~Sd}~ww)a*Execute a SQL command and return a string representing the results. If the statement returns rows, a string of the results is returned. If the statement returns no rows, an empty string is returned. If the statement throws an error, the error message is returned. rwN)rvrf)r,rqrrrxr/r/r0 run_no_throwszSparkSQL.run_no_throw)NNNNNr ) r rrrrrrrrrrrr@)r2r3r4r5r6rr7r )r7r<)rJr3r7r3)rRrr7r3)rAr r7rl)rhrr7rG)rp)rqr3rrr3r7r3)__name__ __module__ __qualname____doc__r1 classmethodr;r'r"rQr]rUrorervryrzr/r/r/r0r s, ?         r N) __future__rtypingrrrrrrrr r r r/r/r/r0s