ˀh?ddlmZddlZddlmZmZmZddlmZddl m Z ddl m Z m Z ddlmZddlmZdd lmZe r-ej(e5dd lmZmZddddd l mZmZmZej(e5dd lmZdddGd dZGddZGddZ ddZ ddZ! d dZ" d!dZ#GddeZ$ d"dZ%GddeZ&GddeZ'y#1swYxYw#1swY}xYw)#) annotationsN)IterableMappingSequence)Path) TYPE_CHECKING) DataFramecol)PartitioningScheme)issue_unstable_warning)Expr) PyDataFramePyExpr)IOAnyCallable)PyPartitioningc@eZdZUdZd dZded<ded<ded<d dZy ) KeyedPartitiona0 A key-value pair for a partition. .. warning:: This functionality is currently considered **unstable**. It may be changed at any point without it being considered a breaking change. See Also -------- PartitionByKey PartitionParted KeyedPartitionContext namestr str_value raw_valuerc.||_||_||_yNrrr)selfrrrs o/var/www/html/wine-match-dev/backend/winematch-backend/venv/lib/python3.12/site-packages/polars/io/partition.py__init__zKeyedPartition.__init__&s ""c8|jd|jS)zGet the `key=value`.=)rr)rs r hive_namezKeyedPartition.hive_name/s))Adnn-..r N)rrrrrrreturnNone)r$r)__name__ __module__ __qualname____doc__r__annotations__r#r rrrs" # INN/r rczeZdZUdZ dd Zded<ded<ded<ded<d ed<d ed <dd Zy )KeyedPartitionContexta* Callback context for a partition creation using keys. .. warning:: This functionality is currently considered **unstable**. It may be changed at any point without it being considered a breaking change. See Also -------- PartitionByKey PartitionParted file_idxintpart_idx in_part_idxkeyslist[KeyedPartition] file_pathr full_pathcX||_||_||_||_||_||_yrr.r0r1r2r4r5)rr.r0r1r2r4r5s rrzKeyedPartitionContext.__init__Bs/!   & ""r ct|jdkDsJt|jdj}|jddD]}|t|jz} |S)z$The keys mapped to hive directories.rN)lenr2rr#)rpkeys r hive_dirszKeyedPartitionContext.hive_dirs[sf499~!!! 1'') *99QR= 'C cmmo& &A 'r N)r.r/r0r/r1r/r2r3r4rr5rr$r%)r$r)r&r'r(r)rr*r=r+r rr-r-4sz ### # # #  ## # MM O r r-c8eZdZUdZd dZded<ded<ded<y) BasePartitionContexta  Callback context for a partition creation. .. warning:: This functionality is currently considered **unstable**. It may be changed at any point without it being considered a breaking change. See Also -------- PartitionMaxSize r.r/r4rr5c.||_||_||_yrr.r4r5)rr.r4r5s rrzBasePartitionContext.__init__qs  ""r N)r.r/r4rr5rr$r%)r&r'r(r)rr*r+r rr?r?ds  # MO r r?cyfdS)Nc t|jt|jt|jS)NrA)r?r.rr4r5)ctx file_path_cbs rz)_cast_base_file_path_cb..s3|\\3==)3==) r r+rEs`r_cast_base_file_path_cbrH}s r cyfdS)NcHt|j|j|j|jDcgc].}t |j |j|j0c}t|jt|jScc}w)Nrr7) r-r.r0r1r2rrrrrr4r5)rDkvrEs rrFz*_cast_keyed_file_path_cb..s~|\\\\ (( BLLBLL 3==)3==)  s3Br+rGs`r_cast_keyed_file_path_cbrLs r c dd}|yt|trt|jgSt|tr |jgSt|t r|Dcgc] }|| c}Sd|}t |cc}w)Nct|trt|jSt|tr |jSd|}t |)N&cannot do a per partition sort by for ) isinstancerr _pyexprr TypeError)vmsgs r prepare_onez3_prepare_per_partition_sort_by..prepare_onesD a q6>> ! 4 99 :1%@CC. r rO)rS str | Exprr$r)rPrr rQr rrR)erUrSrTs r_prepare_per_partition_sort_byrXs|! y As A At  { Ax ()*1 A**6qe<n+sBcydfd }|S)Nc<tj|yr)r _from_pydf)pydffs rcbz$_prepare_finish_callback..cbs )  t $%r )r\rr$r%r+)r]r^s` r_prepare_finish_callbackr_s y& Ir cDeZdZdZdddd dfdZxZS)PartitionMaxSizeao Partitioning scheme to write files with a maximum size. This partitioning scheme generates files that have a given maximum size. If the size reaches the maximum size, it is closed and a new file is opened. .. warning:: This functionality is currently considered **unstable**. It may be changed at any point without it being considered a breaking change. Parameters ---------- base_path The base path for the output files. file_path A callback to register or modify the output path for each partition relative to the `base_path`. The callback provides a :class:`polars.io.partition.BasePartitionContext` that contains information about the partition. If no callback is given, it defaults to `{ctx.file_idx}.{EXT}`. max_size : int The maximum size in rows of each of the generated files. per_partition_sort_by Columns or expressions to sort over within each partition. Note that this might increase the memory consumption needed for each partition. finish_callback A callback that gets called when the query finishes successfully. For parquet files, the callback is given a dataframe with metrics about all files written files. Examples -------- Split a parquet file by over smaller CSV files with 100 000 rows each: >>> pl.scan_parquet("/path/to/file.parquet").sink_csv( ... PartitionMax("./out", max_size=100_000), ... ) # doctest: +SKIP See Also -------- PartitionByKey PartitionParted polars.io.partition.BasePartitionContext N)r4per_partition_sort_byfinish_callbackc tdt| tj|t ||t |t|y)N0partitioning strategies are considered unstable.) base_pathrEmax_sizerbrc)r superrr new_max_sizerHrXr_)rrfr4rgrbrc __class__s rrzPartitionMaxSize.__init__sL QR   ' '#4Y?!&D)'!9 I  r ) rf str | Pathr4ICallable[[BasePartitionContext], Path | str | IO[bytes] | IO[str]] | Nonergr/rb(str | Expr | Iterable[str | Expr] | Nonerc"Callable[[DataFrame], None] | Noner$r%r&r'r(r)r __classcell__rjs@rrarasY.jJN>B       H <    r racdd}t|trt|jg}|St|tr|jg}|St|t r!|Dcgc]}||j}}|St|t r=|jDcgc] \}}|j|j"}}}|Sd}t|cc}wcc}}w)Nc<t|tr t|S|Sr)rPrr )is rto_exprz_lower_by..to_exprs a q6MHr zinvalid `by` type)rtrVr$r ) rPrr rQr rritemsaliasrR)byru lowered_byrWnrTs r _lower_byr{s"c"goo&   B jj\   B !245Qgaj((5 5  B 57XXZ@TQaggaj((@ @ "n 6@s C %CcJeZdZdZddddd dfdZxZS)PartitionByKeya Partitioning scheme to write files split by the values of keys. This partitioning scheme generates an arbitrary amount of files splitting the data depending on what the value is of key expressions. The amount of files that can be written is not limited. However, when writing beyond a certain amount of files, the data for the remaining partitions is buffered before writing to the file. .. warning:: This functionality is currently considered **unstable**. It may be changed at any point without it being considered a breaking change. Parameters ---------- base_path The base path for the output files. Use the `mkdir` option on the `sink_*` methods to ensure directories in the path are created. file_path A callback to register or modify the output path for each partition relative to the `base_path`. The callback provides a :class:`polars.io.partition.KeyedPartitionContext` that contains information about the partition. If no callback is given, it defaults to `{ctx.keys.hive_dirs()}/{ctx.in_part_idx}.{EXT}`. by The expressions to partition by. include_key : bool Whether to include the key columns in the output files. per_partition_sort_by Columns or expressions to sort over within each partition. Note that this might increase the memory consumption needed for each partition. finish_callback A callback that gets called when the query finishes successfully. For parquet files, the callback is given a dataframe with metrics about all files written files. Examples -------- Split into a hive-partitioning style partition: >>> ( ... pl.DataFrame({"a": [1, 2, 3], "b": [5, 7, 9], "c": ["A", "B", "C"]}) ... .lazy() ... .sink_parquet( ... PartitionByKey( ... "./out", ... by=[pl.col.a, pl.col.b], ... include_key=False, ... ), ... mkdir=True, ... ) ... ) # doctest: +SKIP Split a parquet file by a column `year` into CSV files: >>> pl.scan_parquet("/path/to/file.parquet").sink_csv( ... PartitionByKey( ... "./out/", ... file_path=lambda ctx: f"year={ctx.keys[0].str_value}.csv", ... by="year", ... ), ... ) # doctest: +SKIP See Also -------- PartitionMaxSize PartitionParted polars.io.partition.KeyedPartitionContext NTr4 include_keyrbrcc tdt|}t| t j |t |||t|t|yNre)rfrErxrrbrc r r{rhrr new_by_keyrLrXr_ rrfr4rxrrbrcryrjs rrzPartitionByKey.__init__yX QRr]    % %#5i@'&D)'!9 I  r rfrkr4JCallable[[KeyedPartitionContext], Path | str | IO[bytes] | IO[str]] | Nonerx6str | Expr | Sequence[str | Expr] | Mapping[str, Expr]rboolrbrmrcrnr$r%rorqs@rr}r}+shKd JN>B    C    H <    r r}cJeZdZdZddddd dfdZxZS)PartitionPartedab Partitioning scheme to split parted dataframes. This is a specialized version of :class:`PartitionByKey`. Where as :class:`PartitionByKey` accepts data in any order, this scheme expects the input data to be pre-grouped or pre-sorted. This scheme suffers a lot less overhead than :class:`PartitionByKey`, but may not be always applicable. Each new value of the key expressions starts a new partition, therefore repeating the same value multiple times may overwrite previous partitions. .. warning:: This functionality is currently considered **unstable**. It may be changed at any point without it being considered a breaking change. Parameters ---------- base_path The base path for the output files. Use the `mkdir` option on the `sink_*` methods to ensure directories in the path are created. file_path A callback to register or modify the output path for each partition relative to the `base_path`.The callback provides a :class:`polars.io.partition.KeyedPartitionContext` that contains information about the partition. If no callback is given, it defaults to `{ctx.keys.hive_dirs()}/{ctx.in_part_idx}.{EXT}`. by The expressions to partition by. include_key : bool Whether to include the key columns in the output files. per_partition_sort_by Columns or expressions to sort over within each partition. Note that this might increase the memory consumption needed for each partition. finish_callback A callback that gets called when the query finishes successfully. For parquet files, the callback is given a dataframe with metrics about all files written files. Examples -------- Split a parquet file by a column `year` into CSV files: >>> pl.scan_parquet("/path/to/file.parquet").sink_csv( ... PartitionParted("./out", by="year"), ... mkdir=True, ... ) # doctest: +SKIP See Also -------- PartitionMaxSize PartitionByKey polars.io.partition.KeyedPartitionContext NTr~c tdt|}t| t j |t |||t|t|yrrrs rrzPartitionParted.__init__rr rrorqs@rrrsg:B JN>B    C    H <    r r)rErlr$rl)rErr$r)rWrmr$zlist[PyExpr] | None)r]rnr$z$Callable[[PyDataFrame], None] | None)rxrr$z list[PyExpr])( __future__r contextlibcollections.abcrrrpathlibrtypingrpolarsr r polars._typingr polars._utils.unstabler polars.exprr suppress ImportError polars.polarsrrrrrrrr-r?rHrLrXr_rar{r}rr+r rrs@"77 !-9   [ )656)(Z%-,-//:--`2   O  P./2 ) ) F )F R>2g 'g TV (V O 66 --s C3<C?3C<?D