o i+b@sddlmZddlZddlZddlZddlZddlmZmZddl m Z m Z m Z ddl mZddlmZddlmZddlmZdd lmZmZmZdd lmZddlmmZdd lmZmZm Z m!Z!m"Z"m#Z#m$Z$m%Z%dd l&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,dd l-m.Z.m/Z/ddl0m1Z1m2Z2m3Z3m4Z4m5Z5ddl6m7Z7m8Z8m9Z9m:Z:ddl;mZ>ddl?m@Z@ddlAmBZBmCZCddlDmEZEddlFmGZGmHZHddlImJZJddlKmLZLmMZMddlNmOZOmPZPddlQmRZRmSZSddlTmUZUmVZVmWZWmXZXmYZYmZZZm[Z[m\Z\m]Z]m^Z^m_Z_m`Z`maZambZbmcZcmdZdmeZemfZfmgZgddlhmiZiddljmkZkmlZlmmZmddlnmoZoddlpmqZqddlrmsZsdd ltmuZudd!lvmwZwmxZxmyZymzZzm{Z{m|Z|m}Z}m~Z~dd"lmZdd#lmZdd$lmZdd%lmZerddlmZddlZddlZddlZddlZddlZdd&lmZdd'lmZdd(lmZdd)lmZmZmZeZd*ZeGd+d,d,ZGd-d.d.ZGd/d0d0ZGd1d2d2eZGd3d4d4eyeeZGd5d6d6exZGd7d8d8exZGd9d:d:exZGd;d<d<ZGd=d>d>eZGd?d@d@eZdrdEdFZedGdHdIdJZGdKdHdHeZGdLdMdMZedNdOGdPdQdQeZdsdUdVZdtd_d`ZdudddeZGdfdgdgeZedIdhGdidjdjZdvdpdqZdS)w) annotationsN)ABCabstractmethod)CallableIterableIterator) dataclass)cached_property)chain) TracebackType) TYPE_CHECKINGAnyTypeVar)Field) AlwaysFalse AlwaysTrueAndBooleanExpressionEqualToIsNullOr Reference)ResidualEvaluator_InclusiveMetricsEvaluatorbindexpression_evaluatorinclusive_projectionmanifest_evaluator)FileIO load_file_io)DataFileDataFileContentManifestContent ManifestEntry ManifestFile)PARTITION_FIELD_ID_STARTUNPARTITIONED_PARTITION_SPEC PartitionKey PartitionSpec)Schema)DeleteFileIndex InspectTable)LocationProviderload_location_providerMaintenanceTable)INITIAL_SEQUENCE_NUMBER TableMetadata) NameMapping) MAIN_BRANCH SnapshotRef)SnapshotSnapshotLogEntry)UNSORTED_SORT_ORDER SortOrder)AddPartitionSpecUpdateAddSchemaUpdateAddSortOrderUpdate AssertCreateAssertRefSnapshotIdAssertTableUUIDAssignUUIDUpdateRemovePropertiesUpdateSetCurrentSchemaUpdateSetDefaultSortOrderUpdateSetDefaultSpecUpdateSetLocationUpdateSetPropertiesUpdateSetSnapshotRefUpdateTableRequirement TableUpdateUpdatesAndRequirementsUpgradeFormatVersionUpdateupdate_table_metadata) UpdateSchema)ManageSnapshotsUpdateSnapshot_FastAppendFilesUpdateSortOrder UpdateSpecUpdateStatistics)IdentityTransform) EMPTY_DICTIcebergBaseModelIcebergRootModel IdentifierKeyDefaultDict PropertiesRecord TableVersion) strtobool)ExecutorFactory)Config)property_as_bool)DuckDBPyConnectionIcebergDataFusionTable)Catalog)RESTContentFileRESTDeleteFileRESTFileScanTaskz$downcast-ns-timestamp-to-us-on-writec@s*eZdZUdZdZded<dZded<dS) UpsertResultzSummary the upsert operation.rint rows_updated rows_insertedN)__name__ __module__ __qualname____doc__rm__annotations__rnrtrtV/home/ubuntu/transcripts/venv/lib/python3.10/site-packages/pyiceberg/table/__init__.pyrks  rkc@seZdZUdZdZdZdZdZdZdZ dZ dZ d Z d Z d Zd Zd ZdZdZdZdZdZdZdZdZdZdZdZdZdZdZdZdZ dZ!dZ"dZ#d Z$d!Z%d"Z&d#Z'd$Z(e'Z)d%Z*d&Z+d'Z,d(e-d)<d*Z.d+Z/d,Z0d-Z1d.Z2dZ3d/Z4d-Z5d0Z6dZ7d1Z8d2Z9d3Z:d4Z;d S)5TablePropertiesz"write.parquet.row-group-size-bytesizwrite.parquet.row-group-limitizwrite.parquet.page-size-byteszwrite.parquet.page-row-limiti Nzwrite.parquet.dict-size-bytesi zwrite.parquet.compression-codeczstdzwrite.parquet.compression-levelNz$write.parquet.bloom-filter-max-bytesz)write.parquet.bloom-filter-enabled.columnzwrite.target-file-size-bytesi zwrite.avro.compression-codecgzipzwrite.metadata.metrics.defaultz truncate(16)zwrite.metadata.metrics.columnzwrite.summary.partition-limitrzwrite.py-location-provider.implzwrite.object-storage.enabledFz&write.object-storage.partitioned-pathsTzwrite.data.pathzwrite.format.defaultparquetzwrite.metadata.pathzwrite.delete.modez copy-on-writez merge-on-readzschema.name-mapping.defaultzformat-versionr_DEFAULT_FORMAT_VERSIONz!commit.manifest.target-size-bytesiz"commit.manifest.min-count-to-mergedzcommit.manifest-merge.enabledz$write.metadata.previous-versions-maxz*write.metadata.delete-after-commit.enabledz"history.expire.max-snapshot-age-msi̿z$history.expire.min-snapshots-to-keep)/z%Transaction._stage..)validaterrrr)rrrrexisting_requirementsnew_requirementrtrtru_stages zTransaction._stagecCs||||jr ||S)zKCheck if the requirements are met, and applies the updates to the metadata.)rrr)rrrrtrtru_apply6s zTransaction._applyT row_filterstr | BooleanExpressioncase_sensitiveDataScancCst|j|jj||dS)zIMinimal data scan of the table with the current state of the transaction.riorr)rrrr)rrrrtrtru_scanCszTransaction._scanformat_versionr_cCs\|dvr td|||jjkrtd|jjd|||jjkr,|t|dfS|S)zSet the table to a certain version. Args: format_version: The newly set version. Returns: The alter table builder. >r}rzz"Unsupported table format version: zCannot downgrade vz table to vr) ValueErrorrrrrK)rrrtrtruupgrade_table_versionIs   z!Transaction.upgrade_table_version propertiesr]kwargsr cKs*|r|rtd|p |}|t|dfS)aSet properties. When a property is already set, it will be overwritten. Args: properties: The properties set on the table. kwargs: properties can also be pass as kwargs. Returns: The alter table builder. z&Cannot pass both properties and kwargsr)rrrF)rrrrrtrtruset_properties]s zTransaction.set_propertiesN snapshot_idrlref_namestrrmax_ref_age_ms int | Nonemax_snapshot_age_msmin_snapshots_to_keeprJc CsFt||||||df}t||jjvr|jj|jnd|df}||fS)zUpdate a ref to a snapshot. Returns: The updates and requirements for the set-snapshot-ref staged )rrrrrrN)rref)rGr>rrefsr) rrrrrrrrrrtrtru_set_ref_snapshotns zTransaction._set_ref_snapshotpartition_records set[Record]rc s|j}|jfdd|jD}t}|D]-}t}t|D]\}}||dur5tt|||nt t|} t || }q"t ||}q|S)zBuild a filter predicate matching any of the input partition records. Args: partition_records: A set of partition records to match Returns: A predicate matching any of the input partition records. csg|] }|jjqSrt) find_field source_idnamerfieldschemartru sz:Transaction._build_partition_predicate..N) rspecrfieldsrr enumeraterrrrr) rrpartition_specpartition_fieldsexprpartition_recordmatch_partition_expressionpospartition_field predicatertrru_build_partition_predicates      z&Transaction._build_partition_predicatesnapshot_propertiesdict[str, str]branch str | NonerPcCs6t|jjtjtj}|j||d}|r|S|S)zDetermine the append type based on table properties. Args: snapshot_properties: Custom properties to be added to the snapshot summary Returns: Either a fast-append or a merge-append snapshot producer. r r) rcrrrvrrupdate_snapshot merge_append fast_append)rr rmanifest_merge_enabledrrtrtru_append_snapshot_producers z%Transaction._append_snapshot_producerallow_incompatible_changesrMcCst||||jdS),Create a new UpdateSchema to alter the columns of this table. Args: allow_incompatible_changes: If changes are allowed that might break downstream consumers. case_sensitive: If field names are case-sensitive. Returns: A new UpdateSchema. )rr name_mapping)rMrrrrrrtrtru update_schemas  zTransaction.update_schemarRcCs t||dS)zCreate a new UpdateSortOrder to update the sort order of this table. Args: case_sensitive: If field names are case-sensitive. Returns: A new UpdateSortOrder. rrQrrrtrtruupdate_sort_orders zTransaction.update_sort_orderrOcCst||jj||dS)zCreate a new UpdateSnapshot to produce a new snapshot for the table. Returns: A new UpdateSnapshot )rrr )rOrr)rr rrtrtrurszTransaction.update_snapshotrVcCs t|dS)z Create a new UpdateStatistics to update the statistics of the table. Returns: A new UpdateStatistics  transactionrUrrtrtruupdate_statisticss zTransaction.update_statisticsdfpa.Tablec Cszddl}Wnty}ztd|d}~wwddlm}m}t||js-td|t t p4d}||j |j ||j j d|j||d0} |jddkrjt||j | j||jjd } | D]} | | qbWddSWddS1s}wYdS) aM Shorthand API for appending a PyArrow table to a table transaction. Args: df: The Arrow dataframe that will be appended to overwrite the table snapshot_properties: Custom properties to be added to the snapshot summary branch: Branch Reference to run the append operation rN(For writes PyArrow needs to be installed _check_pyarrow_schema_compatible_dataframe_to_data_filesExpected PyArrow table, got: Fprovided_schemadowncast_ns_timestamp_to_usrrr write_uuidr!r)pyarrowModuleNotFoundErrorpyiceberg.io.pyarrowr%r& isinstancerrrbget_bool$DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITErrrrshapelist commit_uuidrrappend_data_file) rr!r rpaer%r&r* append_files data_files data_filertrtruappends:    "zTransaction.appendc Csvzddl}Wnty}ztd|d}~wwddlm}m}t||js-td||j r8td|j j D]}t|j t sMtd|q>ttpUd} ||j|j| |jjd |jddkrmdSt} t||jj| ||jjd } d d | D} |j| d } |j| ||d|j||d}| |_| D]}||qWddS1swYdS)aK Shorthand for overwriting existing partitions with a PyArrow table. The function detects partition values in the provided arrow table using the current partition spec, and deletes existing partitions matching these values. Finally, the data in the table is appended to the table. Args: df: The Arrow dataframe that will be used to overwrite the table snapshot_properties: Custom properties to be added to the snapshot summary branch: Branch Reference to run the dynamic partition overwrite operation rNr#r$r'z9Cannot apply dynamic overwrite on an unpartitioned table.zsFor now dynamic overwrite does not support a table with non-identity-transform field in the latest partition spec: Fr(r,cSsh|]}|jqSrt) partition)rr<rtrtrurIsz:Transaction.dynamic_partition_overwrite..)r) delete_filterr rr+)r.r/r0r%r&r1rrrris_unpartitionedr transformrWrbr2r3rrr4uuiduuid4r5rrrr deleterr6r7)rr!r rr8r9r%r&rr*append_snapshot_commit_uuidr;partitions_to_overwriter?r:r<rtrtrudynamic_partition_overwritesV      "z'Transaction.dynamic_partition_overwriteoverwrite_filterBooleanExpression | strc Cs zddl}Wnty}ztd|d}~wwddlm}m} t||js-td|t t p4d} ||j |j | |j j d|tkrQ|j||||d|j||d .} |jddkrv| |j | j||jjd } | D]} | | qnWddSWddS1swYdS) a Shorthand for adding a table overwrite with a PyArrow table to the transaction. An overwrite may produce zero or more snapshots based on the operation: - DELETE: In case existing Parquet files can be dropped completely. - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter. - APPEND: In case new data is being inserted into the table. Args: df: The Arrow dataframe that will be used to overwrite the table overwrite_filter: ALWAYS_TRUE when you overwrite all the data, or a boolean expression in case of a partial overwrite snapshot_properties: Custom properties to be added to the snapshot summary case_sensitive: A bool determine if the provided `overwrite_filter` is case-sensitive branch: Branch Reference to run the overwrite operation rNr#r$r'Fr(r?rr rr+r,)r.r/r0r%r&r1rrrbr2r3rrrrrDrr4r6rrr7)rr!rHr rrr8r9r%r&r*r:r;r<rtrtru overwriteRsD     "zTransaction.overwriter?c Cs"ddlm}m}m}|jjtjtj tj krt j dddt |tr't|}|j||d}|||Wdn1sAwY|jdurt|j||} || |j} |j||d } |durl| |} | } t} td}g}| D]F}||j|jj|jtd j |gd }|!| }t"|dkr|#|j$gfq}t"|t"|kr|#|j$t%||jj||j| |d fq}t"|dkr|j||d&!}| |_'|D]\}}|(||D]}|)|qqWdn1swY|j*s |jst j d dddSdSdS)a Shorthand for deleting record from a table. A delete may produce zero or more snapshots based on the operation: - DELETE: In case existing Parquet files can be dropped completely. - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the delete filter. Args: delete_filter: A boolean expression to delete rows from a table snapshot_properties: Custom properties to be added to the snapshot summary case_sensitive: A bool determine if the provided `delete_filter` is case-sensitive branch: Branch Reference to run the delete operation r) ArrowScanr&$_expression_to_complementary_pyarrowzAMerge on read is not yet supported, falling back to copy-on-writerz stacklevelrNT)rr)rrprojected_schemar)tasks)rr!rr-counterz*Delete operation did not match any records)+r0rLr&rMrrgetrvrrrwarningswarnr1r_parse_row_filterrrDdelete_by_predicaterewrites_neededrrruse_ref plan_filesrBrC itertoolscountrrrto_tablefilterlenr=filer5rKr6delete_data_filer7files_affected)rr?r rrrLr&rMdelete_snapshotbound_delete_filterpreserve_row_filter file_scanfilesr6rRreplaced_files original_filer! filtered_dfoverwrite_snapshotoriginal_data_filereplaced_data_filesreplaced_data_filertrtrurDs           zTransaction.delete join_colslist[str] | Nonewhen_matched_update_allwhen_not_matched_insert_allrkc Cs\zddl}Wnty} ztd| d} ~ wwddlm} ddlm} |durKg}|jjD]} |j | } | durD| | q0t d|t |dkrUt d|s]|s]t d| ||rgt d dd lm}ttptd }||j|j||jjd | ||}t|j|jj||d }||jjvr||}|}g}g}|}|D]D}|j|g}|r| |||}t |dkr| ||}| || ||r| ||}t|j||d}| |}||}qd}d}|r||}t |}|j |t |dkrt!|n|d||d|r(t |}|r(|j |||dt"||dS)LShorthand API for performing an upsert to an iceberg table. Args: df: The input dataframe to upsert with the table's data. join_cols: Columns to join on, if not provided, it will use the identifier-field-ids. when_matched_update_all: Bool indicating to update rows that are matched but require an update due to a value in a non-key column changing when_not_matched_insert_all: Bool indicating new rows to be inserted that do not match any existing rows in the table case_sensitive: Bool indicating if the match should be case-sensitive branch: Branch Reference to run the upsert operation snapshot_properties: Custom properties to be added to the snapshot summary To learn more about the identifier-field-ids: https://iceberg.apache.org/spec/#identifier-field-ids Example Use Cases: Case 1: Both Parameters = True (Full Upsert) Existing row found → Update it New row found → Insert it Case 2: when_matched_update_all = False, when_not_matched_insert_all = True Existing row found → Do nothing (no updates) New row found → Insert it Case 3: when_matched_update_all = True, when_not_matched_insert_all = False Existing row found → Update it New row found → Do nothing (no inserts) Case 4: Both Parameters = False (No Merge Effect) Existing row found → Do nothing New row found → Do nothing (Function effectively does nothing) Returns: An UpsertResult class (contains details of rows updated and inserted) rNr#)expression_to_pyarrow) upsert_utilzField-ID could not be found: zWJoin columns could not be found, please set identifier-field-ids or pass in explicitly.z$no upsert options selected...exitingzSDuplicate rows found in source dataset based on the key columns. No upsert executed)r%Fr(rrr})rHrr )rr )rmrn)#r.r/r0rtpyiceberg.tablerurridentifier_field_idsfind_column_namer=rr_has_duplicate_rowsr%rbr2r3rcreate_match_filterrrrrrYto_arrow_batch_readerr from_batchesget_rows_to_updaterr^ concat_tablesrKrrk)rr!rorqrrrrr r8r9rtrufield_idcolr%r*matched_predicate#matched_iceberg_record_batches_scanmatched_iceberg_record_batchesbatches_to_overwriteoverwrite_predicatesrows_to_insertbatchrowsrows_to_updateoverwrite_mask_predicate expr_matchexpr_match_boundexpr_match_arrowupdate_row_cntinsert_row_cntrtrtruupserts0                   zTransaction.upsert file_paths list[str]check_duplicate_filesc Cst|tt|krtd|r:ddlm}|d|}dd|jj | D}|r:tdd ||j durR|jd itj|j ji|j||d }t|j ||jjd } | D]} || qfWddS1sywYdS) a Shorthand API for adding files as data files to the table transaction. Args: file_paths: The list of full file paths to be added as data files to the table Raises: FileNotFoundError: If the file does not exist. ValueError: Raises a ValueError given file_paths contains duplicate files ValueError: Raises a ValueError given file_paths already referenced by table zFile paths must be uniquerN file_pathcSsg|]}|dqS)rrt)rr`rtrtrurrz)Transaction.add_files..z>Cannot add files that are already referenced by table, files: , r+)rrrrt)r_setrpyarrow.computecomputerisinrinspectr;r^ to_pylistjoinrrrrvrrmodel_dump_jsonr_parquet_files_to_data_filesrr7) rrr rrpcrreferenced_filesr:r;r<rtrtru add_filess(    "zTransaction.add_filesrTcCt|S)zyCreate a new UpdateSpec to update the partitioning of the table. Returns: A new UpdateSpec. rSrrtrtru update_speczTransaction.update_specremovalscGs|t|dfS)zRemove properties. Args: removals: Properties to be removed. Returns: The alter table builder. )r)rrA)rrrtrtruremove_propertiess zTransaction.remove_propertieslocationcCtd)zSet the new table location. Args: location: The new location of the table. Returns: The alter table builder. zNot yet implementedNotImplementedError)rrrtrtruupdate_location zTransaction.update_locationcCsNt|jdkr|jt|jjdf7_|jj|j|jdd|_d|_|jS)zmCommit the changes to the catalog. Returns: The table with the updates applied. rrBrrrt)r_rrr?r table_uuidr _do_commitrrtrtrurszTransaction.commit_transaction)F)rrrr)rr2rr)rrrrrrrr)rt)rrrrrr)rrrrrr)rr_rr)rr]rr rr)NNN)rrlrrrrrrrrrrrrJ)rrrr)r r rrrrPFTrrrrrrMTrrrrR)r r rrrrOrrVr!r"r r rrrr r!r"rHrIr r rrrrrr) r?rr r rrrrrrr!r"rorprqrrrrrrrrr r rrk rrr r rrrrrr)rrT)rrrr)rrrrrr)"rorprqrsrpropertyrrrrr ALWAYS_TRUErrrXrrr r4rrrrr r=rGrKrDrrrrrrrtrtrtrursr         !    ' B @ h  )   rcs6eZdZdZdddZdfd d Zdd dZZS)CreateTableTransactionz8A transaction that involves the creation of a new table.rr2rrcCs |jt|jdt|jdf7_|}|jt|dtddf7_|}| r8|jt t df7_n |jt |df7_|jt ddf7_| |j}|dus[|jrg|jttd f7_n |jt|d f7_|jtdd f7_|jt|jd t|jd f7_dS) zqSet the initial changes that can reconstruct the initial table metadata when creating the CreateTableTransaction.rr)schema_ schema_id)rspec_idN) sort_order) sort_order_id)rr)rr@rrKrrr;rBrr@r:r&rDsort_order_by_iddefault_sort_order_id is_unsortedr<r8rCrErrFr)rrrrrrtrtru_initial_changess.      z'CreateTableTransaction._initial_changesr StagedTablecs tj|dd||jdS)NFr)superrrr)rr __class__rtrurszCreateTableTransaction.__init__rcCs6t|jdkr|jj|jtfdd|_d|_|jS)zCommit the changes to the catalog. In the case of a CreateTableTransaction, the only requirement is AssertCreate. Returns: The table with the updates applied. rrrt)r_rrrr=rrrtrtrur sz)CreateTableTransaction.commit_transaction)rr2rr)rrr)rorprqrrrrr __classcell__rtrtrrurs   rc@s&eZdZUdZedddZded<dS) Namespacez/Reference to one or more levels of a namespace..z.Reference to one or more levels of a namespace) descriptionrrootN)rorprqrrrrrsrtrtrtrurs rc@s"eZdZUdZded<ded<dS)TableIdentifierz&Fully Qualified identifier to a table.r namespacerrN)rorprqrrrsrtrtrtrur's  rc@sDeZdZUdZeZded<eedZded<eedZ ded<d S) CommitTableRequestz0A pydantic BaseModel for a table commit request.r identifier)default_factoryrrrrN) rorprqrrrrrstuplerrrtrtrtrur.s rc@s,eZdZUdZded<eddZded<dS) CommitTableResponsez1A pydantic BaseModel for a table commit response.r2rzmetadata-location)aliasrmetadata_locationN)rorprqrrrsrrrtrtrtrur6s rc@seZdZUdZeZded<ded<eZded<ded <d ed <d ed <efdddZ dddZ e dddZ e dddZ dddZdddZed d!d"ed"fdd/d0Ze dd2d3Zdd5d6Zdd8d9Zdd;d<Zdd>d?ZddAdBZddDdEZddGdHZe ddIdJZddKdLZddNdOZe ddPdQZddSdTZddVdWZddXdYZ dd[d\Z!ddd_d`Z"ddbdcZ#ddedfZ$ddhdiZ%dddmdnZ&dddpdqZ'ddsdtZ(d"d!d!d!e)efdddZ*ee)fdddZ+ee)fdddZ,eed!e)fdddZ-eed!e)fdddZ.ed!e)fdddZ/ddddZ0dddZ1e2dddZ3dddZ4dddZ5dddZ6dddZ7dddZ8dddZ9dddZ:d"S)rzAn Iceberg table.r[ _identifierr2rrrrrrgcatalogr configrrrcCs(||_||_||_||_||_||_dSr)rrrrrr)rrrrrrrrtrtrurGs   zTable.__init__rcCr)zCreate a new transaction object to first stage the changes, and then commit them to the catalog. Returns: The transaction object )rrrtrtrurWrzTable.transactionr,cCr)zReturn the InspectTable object to browse the table metadata. Returns: InspectTable object based on this Table. r+rrtrtrur_z Table.inspectr0cCr)zReturn the MaintenanceTable object for maintenance. Returns: MaintenanceTable object based on this Table. r/rrtrtru maintenancehrzTable.maintenancecCs:|j|j}||j|j|j|_|j|_|j|_|S)zxRefresh the current table metadata. Returns: An updated instance of the same Iceberg table )r load_tabler _check_uuidrrr)rfreshrtrtrurefreshqs z Table.refreshcCs|jS)zqReturn the identifier of this table. Returns: An Identifier tuple of the table name )rrrtrtrur~sz Table.name*TNrrselected_fieldstuple[str, ...]rrrroptionsr]limitrc Cs$t|j|j|||||||j|jd S)awFetch a DataScan based on the table's current metadata. The data scan can be used to project the table's data that matches the provided row_filter onto the table's current schema. Args: row_filter: A string or BooleanExpression that describes the desired rows selected_fields: A tuple of strings representing the column names to return in the output dataframe. case_sensitive: If True column matching is case sensitive snapshot_id: Optional Snapshot ID to time travel to. If None, scans the table as of the current snapshot ID. options: Additional Table properties as a dictionary of string key value pairs to use for this scan. limit: An integer representing the number of rows to return in the scan result. If None, fetches all matching rows. Returns: A DataScan based on the table's current metadata. ) rrrrrrrrrtable_identifier)rrrrrrrrrrrrrtrtruscans&z Table.scanr_cC|jjSr)rrrrtrtrurzTable.format_versionr)ctfddjjDS)z!Return the schema for this table.c3"|] }|jjjkr|VqdSr)rrcurrent_schema_idrrrrtru  zTable.schema..)nextrschemasrrtrrurz Table.schemadict[int, Schema]cCdd|jjDS)z*Return a dict of the schema of this table.cSi|]}|j|qSrtrrrtrtru rz!Table.schemas..)rrrrtrtrurz Table.schemasr(cr)z(Return the partition spec of this table.c3rr)rrdefault_spec_idrrrrtrurrzTable.spec..)rrpartition_specsrrtrrurrz Table.specdict[int, PartitionSpec]cCr)z-Return a dict the partition specs this table.cSrrtrrrtrtrurrzTable.specs..)rrrrtrtruspecsrz Table.specsr9cr)z$Return the sort order of this table.c3rr)order_idrrrrrrtrursz#Table.sort_order..)rr sort_ordersrrtrrurs zTable.sort_orderdict[int, SortOrder]cCr)z/Return a dict of the sort orders of this table.cSrrt)rrrtrtrurrz%Table.sort_orders..)rrrrtrtrurrzTable.sort_ordersrlcCs|jjr|jjStdS)zmReturn the highest assigned partition field ID across all specs or 999 if only the unpartitioned spec exists.r})rlast_partition_idr%rrtrtrurszTable.last_partition_idcCr)zProperties of the table.)rrrrtrtrurszTable.propertiescCr)z!Return the table's base location.)rrrrtrtrurrzTable.locationr-cCst|jj|jjdS)z%Return the table's location provider.)table_locationtable_properties)r.rrrrrtrtrulocation_providerszTable.location_providercCrr)rlast_sequence_numberrrtrtrur rzTable.last_sequence_numberSnapshot | NonecCs|jjdur ||jjSdS)zQGet the current snapshot for this table, or None if there is no current snapshot.N)rcurrent_snapshot_idsnapshot_by_idrrtrtrucurrent_snapshots zTable.current_snapshotlist[Snapshot]cCrr)r snapshotsrrtrtrurzTable.snapshotscCs |j|S)z[Get the snapshot of this table with the given id, or None if there is no matching snapshot.)rr)rrrtrtrurs zTable.snapshot_by_idrcCs"|jj|}r||jSdS)zUReturn the snapshot referenced by the given name or null if no such reference exists.N)rrrSrr)rrrrtrtrusnapshot_by_names zTable.snapshot_by_name timestamp_ms inclusivecCs>t|D]}|r|j|ks|j|kr||jSqdS)aCGet the snapshot that was current as of or right before the given timestamp, or None if there is no matching snapshot. Args: timestamp_ms: Find snapshot that was current at/before this timestamp inclusive: Includes timestamp_ms in search when True. Excludes timestamp_ms when False N)reversedhistoryrrr)rrr log_entryrtrtrusnapshot_as_of_timestamps zTable.snapshot_as_of_timestamplist[SnapshotLogEntry]cCr)z'Get the snapshot history of this table.)r snapshot_logrrtrtrur rz Table.historyrNcCtt|dddS)a* Shorthand to run snapshot management operations like create branch, create tag, etc. Use table.manage_snapshots().().commit() to run a specific operation. Use table.manage_snapshots().().().commit() to run multiple operations. Pending changes are applied on commit. We can also use context managers to make more changes. For example, with table.manage_snapshots() as ms: ms.create_tag(snapshot_id1, "Tag_A").create_tag(snapshot_id2, "Tag_B") Trr)rNrrrtrtrumanage_snapshotss zTable.manage_snapshotsrVcCr)a_ Shorthand to run statistics management operations like add statistics and remove statistics. Use table.update_statistics().().commit() to run a specific operation. Use table.update_statistics().().().commit() to run multiple operations. Pending changes are applied on commit. We can also use context managers to make more changes. For example: with table.update_statistics() as update: update.set_statistics(statistics_file=statistics_file) update.remove_statistics(snapshot_id=2) Trr)rVrrrtrtrur  szTable.update_statisticsFrrMcCstt|dd|||dS)rTr)rrrr)rMrrrrtrtrur1s  zTable.update_schemarRcCtt|dd|dS)zCreate a new UpdateSortOrder to update the sort order of this table. Returns: A new UpdateSortOrder. Tr)rr)rRrrrtrtrurBzTable.update_sort_orderNameMapping | NonecCs |jS)z(Return the table's field-id NameMapping.)rrrrtrtrurJ zTable.name_mappingr!r"rorprqrrrrr rkc CsD|}|j|||||||dWdS1swYdS)rs)r!rorqrrrrr N)rr) rr!rorqrrrrr txrtrtrurNs 0$z Table.upsertcC>|}|j|||dWddS1swYdS)aC Shorthand API for appending a PyArrow table to the table. Args: df: The Arrow dataframe that will be appended to overwrite the table snapshot_properties: Custom properties to be added to the snapshot summary branch: Branch Reference to run the append operation r!r rN)rr=rr!r rr#rtrtrur=s "z Table.appendcCr$)aShorthand for dynamic overwriting the table with a PyArrow table. Old partitions are auto detected and replaced with data files created for input arrow table. Args: df: The Arrow dataframe that will be used to overwrite the table snapshot_properties: Custom properties to be added to the snapshot summary branch: Branch Reference to run the dynamic partition overwrite operation r%N)rrGr&rtrtrurGs "z!Table.dynamic_partition_overwriterHrIcCsB|}|j|||||dWddS1swYdS)a Shorthand for overwriting the table with a PyArrow table. An overwrite may produce zero or more snapshots based on the operation: - DELETE: In case existing Parquet files can be dropped completely. - OVERWRITE: In case existing Parquet files need to be rewritten to drop rows that match the overwrite filter.. - APPEND: In case new data is being inserted into the table. Args: df: The Arrow dataframe that will be used to overwrite the table overwrite_filter: ALWAYS_TRUE when you overwrite all the data, or a boolean expression in case of a partial overwrite snapshot_properties: Custom properties to be added to the snapshot summary case_sensitive: A bool determine if the provided `overwrite_filter` is case-sensitive branch: Branch Reference to run the overwrite operation )r!rHrr rN)rrK)rr!rHr rrr#rtrtrurKs "zTable.overwriter?cCs@|}|j||||dWddS1swYdS)a Shorthand for deleting rows from the table. Args: delete_filter: The predicate that used to remove rows snapshot_properties: Custom properties to be added to the snapshot summary case_sensitive: A bool determine if the provided `delete_filter` is case-sensitive branch: Branch Reference to run the delete operation rJN)rrD)rr?r rrr#rtrtrurDs "z Table.deleterrrcCs@|}|j||||dWddS1swYdS)a Shorthand API for adding files as data files to the table. Args: file_paths: The list of full file paths to be added as data files to the table Raises: FileNotFoundError: If the file does not exist. )rr rrN)rr)rrr rrr#rtrtrurs "zTable.add_filesrTcCr)NTrr)rTrrrtrtrurszTable.update_specdict[str, SnapshotRef]cCr)z,Return the snapshot references in the table.)rrrrtrtrurrz Table.refscurrent_metadata new_metadatacCs,|j}|j}||krtd|d|dS)z3Validate that the table UUID matches after refresh.z#Table UUID does not match: current=z != refreshed=N)rr)r(r)current refreshedrtrtrurs zTable._check_uuidrrrrc Cs|j|||}||j|jz |j|j|j|jWnty9}ztjd|ddWYd}~nd}~ww|j|_|j |_ dS)Nz,Failed to delete old metadata after commit: rzrN) r commit_tablerr_delete_old_metadatar ExceptionrTrUr)rrrresponser9rtrtrurs  zTable._do_commitotherr cCs6t|tr||ko|j|jko|j|jkSdS)z8Return the equality of two instances of the Table class.F)r1rrrr)rr0rtrtru__eq__s (z Table.__eq__c sjj}dfddjD}ddfddjDd}ddfd djDd}d rGt nd }|d |d |d|d| }|S)z4Return the string representation of the Table class.z, c3 |] }rt|VqdSr)rr)rcolumnrrtrurz!Table.__repr__..zpartition by: [rc3s|] }r|jVqdSr)rrrrrtrur s]z sort order: [c3r2r)rrrrrtrur!r4z snapshot: nullz( z ), z, ) rtable_name_fromrrrcolumnsrrrrr)r table_name schema_str partition_strsort_order_str snapshot_str result_strrtrru__repr__s&& zTable.__repr__daft.DataFramecCddl}||S)zRead a Daft DataFrame lazily from this Iceberg table. Returns: daft.DataFrame: Unmaterialized Daft Dataframe created from the Iceberg table rN)daft read_iceberg)rrBrtrtruto_daft& z Table.to_daft bd.DataFramecCsddlm}||S)zRead a bodo DataFrame lazily from this Iceberg table. Returns: bd.DataFrame: Unmaterialized Bodo Dataframe created from the Iceberg table rN) bodo.pandaspandasread_iceberg_table)rbdrtrtruto_bodo0s  z Table.to_bodo pl.LazyFramecCrA)zLazily read from this Apache Iceberg table. Returns: pl.LazyFrame: Unmaterialized Polars LazyFrame created from the Iceberg table rN)polars scan_iceberg)rplrtrtru to_polars:rEzTable.to_polarsrfcCs(ddlm}|||j|jjdS)aXReturn the DataFusion table provider PyCapsule interface. To support DataFusion features such as push down filtering, this function will return a PyCapsule interface that conforms to the FFI Table Provider required by DataFusion. From an end user perspective you should not need to call this function directly. Instead you can use ``register_table`` in the DataFusion SessionContext. Returns: A PyCapsule DataFusion TableProvider interface. Example: ```python from datafusion import SessionContext from pyiceberg.catalog import load_catalog import pyarrow as pa catalog = load_catalog("catalog", type="in-memory") catalog.create_namespace_if_not_exists("default") data = pa.table({"x": [1, 2, 3], "y": [4, 5, 6]}) iceberg_table = catalog.create_table("default.test", schema=data.schema) iceberg_table.append(data) ctx = SessionContext() ctx.register_table("test", iceberg_table) ctx.table("test").show() ``` Results in ``` DataFrame() +---+---+ | x | y | +---+---+ | 1 | 4 | | 2 | 5 | | 3 | 6 | +---+---+ ``` rre)rrfile_io_properties)pyiceberg_core.datafusionrfrrrr__datafusion_table_provider__)rrfrtrtrurSDs %z#Table.__datafusion_table_provider__)rr[rr2rrrrrrgrr rrr)rr,)rr0r)rr[rrrrrrrrrr]rrrr)rr_rr))rr)rr()rr)rr9)rrrrl)rr )rr)rr-rr )rr)rrlrr )rrrr r)rrlrrrr )rr)rrNrrrr)rr!rrr) r?rIr r rrrrrrr)rrrrT)rr')r(r2r)r2rr)rrrrrr)r0r rrrr@)rrF)rrL)rrf);rorprqrrrrrsrrXrrrrrrrrrrrrrrrrrrrr r rrrrrrrr rrrr4rr=rGrKrDrrr staticmethodrrr1r?rDrKrPrSrtrtrtrur=s        3                       ;  $         rc@s>eZdZdZdddZeefdd d Zeefdd d ZdS) StaticTablezKLoad a table directly from a metadata file (i.e., without using a catalog).rrcCr)z#Refresh the current table metadata.zTo be implementedrrrtrtrururzStaticTable.refreshrrrr]cCstj|dd}t||d}||}|}|d}Wdn1s)wY|dr;tj|d|S| rKtj|dd|dStj|d|dS)Nrzversion-hint.textrrzutf-8.metadata.jsonv) ospathrr new_inputopenreaddecodeendswith isnumeric)clsrrversion_hint_locationrr`streamcontentrtrtru$_metadata_location_from_version_hintys    z0StaticTable._metadata_location_from_version_hintcCsv|ds t||}t||d}||}ddlm}||}ddlm }|d|f||ti||j |ddS)Nr\r[r) FromInputFile) NoopCatalogz static-table)rrrrr) rdrZrjrr`pyiceberg.serializersrkrpyiceberg.catalog.nooprlr)rfrrrr`rkrrlrtrtru from_metadatas       zStaticTable.from_metadataNr)rrrr]rr)rrrr]rrZ) rorprqrrr classmethodrXrjrortrtrtrurZrs rZc@s8eZdZdddZedddedfdddZdddZdS)rrrcCr)NzCannot refresh a staged tablerrrtrtrurrzStagedTable.refreshrTNrrrrrrrrrr]rrcCr)NzCannot scan a staged tablerqrrtrtrurrzStagedTable.scanr@cCr)Nz1Cannot convert a staged table to a Daft DataFramerqrrtrtrurDrzStagedTable.to_daftrrTrX)rorprqrrrXrrDrtrtrtrurs   rrrrrcCst|tr t|S|S)aAccept an expression in the form of a BooleanExpression or a string. In the case of a string, it will be converted into a unbound BooleanExpression. Args: expr: Expression as a BooleanExpression or a string. Returns: An unbound BooleanExpression. )r1rparserparse)rrtrtrurVs rVS TableScanT)bound covariantc@seZdZUded<ded<ded<ded<d ed <d ed <d ed<d ed<ded<ded<edddedddfdCddZdDddZdEdd Ze dFd"d#Z e dGd%d&Z e dHd*d+Z e dId-d.Z dJd2d3ZdKd6d7ZdLd9d:ZdMdd?Ze dPdAdBZdS)Qrur2rrrrrrrrrrrr]rrCatalog | NonerIdentifier | NonerrTNrc CsD||_||_t||_||_||_||_||_||_| |_ | |_ dSr) rrrVrrrrrrrr) rrrrrrrrrrrrtrtrurs   zTableScan.__init__rr cCs|jr |j|jS|jSr)rrrrrrtrtrusnapshots zTableScan.snapshotr)cs|j}|jdurG|j|jdur?jdur>ztfdd|jjD}Wnty=tj djddYn wnt d|jd|j vrN|S|j |j d|j iS) Nc3s |] }|jjkr|VqdSrrrrzrtrursz'TableScan.projection..z*Metadata does not contain schema with id: rzrNzSnapshot not found: rr)rrrrrrr StopIterationrTrUrrselectr)rcurrent_schemartr{ru projections"      zTableScan.projectionIterable[ScanTask]cCdSrrtrrtrtrurZzTableScan.plan_filesr"cCrrrtrrtrtruto_arrowrzTableScan.to_arrowrr pd.DataFramecKrrrtrrrtrtru to_pandasrzTableScan.to_pandas pl.DataFramecCrrrtrrtrtrurP rzTableScan.to_polarsrrt overridesc sRddlm}|tjjdh}fdd|D}tdii||S)z5Create a copy of this table scan with updated fields.r) signaturercsi|]}|t|qSrt)getattr)rparamrrtrursz$TableScan.update..Nrt)rrrr parameterskeys)rrrparamsrrtrruupdate s zTableScan.updaterrcCsB|jr td|j|j|}r|j|jdStd|)Nz-Cannot override ref, already set snapshot id=)rzCannot scan unknown ref=)rrrrr)rrrzrtrtrurYs zTableScan.use_ref field_namescGs6d|jvr |j|dS|jtt|jt|dS)Nr)r)rrrr intersection)rrrtrtrur}s   zTableScan.selectrcCs|jt|jt|dS)N)r)rrrrV)rrrtrtrur^$szTableScan.filtercCs |j|dS)Nr)rrrtrtruwith_case_sensitive's zTableScan.with_case_sensitiverlcCrrrtrrtrtrur\*rzTableScan.count)rr2rrrrrrrrrrrr]rrrrxrryrWrU)rrrr"rr rrrr)rrtrr rrt)rrtrrrrt)rrtrrrrt)rrtrrrrtr)rrtrrrrtrV)rorprqrsrrXrrzrrrZrrrPrrYr}r^rr\rtrtrtrurusJ           c@s eZdZdS)ScanTaskN)rorprqrtrtrtrur.srF)initc@sHeZdZUdZded<ded<ded<defdd dZedddZdS) FileScanTaskzATask representing a data file and its corresponding delete files.r r`z set[DataFile] delete_filesrresidualNr<set[DataFile] | NonerrcCs||_|pt|_||_dSr)r`rrr)rr<rrrtrtrur:s  zFileScanTask.__init__ rest_taskrjlist[RESTDeleteFile]cCs|ddlm}t|j}t}|jr/|jD]}||}t||r'td|j| t|qt |||j r:|j dSt dS)axConvert a RESTFileScanTask to a FileScanTask. Args: rest_task: The REST file scan task. delete_files: The list of delete files from the ScanTasks response. Returns: A FileScanTask with the converted data and delete files. Raises: NotImplementedError: If equality delete files are encountered. r)RESTEqualityDeleteFilez1PyIceberg does not yet support equality deletes: )r<rr) $pyiceberg.catalog.rest.scan_planningr_rest_file_to_data_filer<rdelete_file_referencesr1rraddrresidual_filterr)rrrr<resolved_deletesidx delete_filertrtrufrom_rest_responseDs     zFileScanTask.from_rest_response)r<r rrrrrr)rrjrrrr) rorprqrrrsrrrYrrtrtrtrur2s   r rest_filerhr cCsddlm}t||r4|jr|jnd}|jr|jnd}|jr'|jnd}|jr1|jnd}nd}d}d}d}tj t |j |j |j|jrOt|jnt|j|j|||||j|jd }|j|_|S)z3Convert a REST content file to a manifest DataFile.r) RESTDataFileN) rir file_formatr> record_countfile_size_in_bytes column_sizes value_countsnull_value_countsnan_value_counts split_offsetsr)rrr1rto_dictrrrr from_argsr!from_rest_typerirrr>r^rrrrr)rrrrrrr<rtrtrurhs4   rrrmanifestr$partition_filterCallable[[DataFile], bool]metrics_evaluatorlist[ManifestEntry]csfdd|j|ddDS)zOpen a manifest file and return matching manifest entries. Returns: A list of ManifestEntry that matches the provided filters. cs$g|]}|jr|jr|qSrtr<)rmanifest_entryrrrtrursz"_open_manifest..T)discard_deleted)fetch_manifest_entry)rrrrrtrru_open_manifests r manifestslist[ManifestFile]rlcCs,z tdd|DWStytYSw)Ncss&|]}|jtjkr|jptVqdSr)rir"DATAmin_sequence_numberr1rrrtrtrurs z'_min_sequence_number..)minrr1)rrtrtru_min_sequence_numbers rc@seZdZdBddZedCdd ZdDd d ZdEddZdFddZdGddZ e dHddZ dIddZ dJddZ dKd!d"ZdKd#d$ZdKd%d&ZdLd(d)ZdMd+d,ZdNd0d1ZdOdPd8d9ZdQd;d<ZdRd>d?ZdSd@dAZd2S)TrrrlrrcCs(t|j|j||j}||jSr)rrrrrr)rrprojectrtrtru_build_partition_projections z$DataScan._build_partition_projection&KeyDefaultDict[int, BooleanExpression]cCs t|jSr)r\rrrtrtrupartition_filtersr"zDataScan.partition_filtersCallable[[ManifestFile], bool]cCs*|j|}t||j|j||jSr)rrrrrr)rrrrtrtru_build_manifest_evaluatorsz"DataScan._build_manifest_evaluatorrcsBj|}|j}t|jj|fddS)Ncstj|jSr)rrr>rpartition_exprpartition_schemarrtrurz5DataScan._build_partition_evaluator..)rrpartition_typerr)rr)rrrrrtrru_build_partition_evaluators   z#DataScan._build_partition_evaluatorcs,jtjddfddS)Ninclude_empty_filesfalsecstjj|Sr)rrrevalrrrrrtrursz3DataScan._build_metrics_evaluator..)rrr`rrSrrtrru_build_metrics_evaluators z!DataScan._build_metrics_evaluator'Callable[[DataFile], ResidualEvaluator]cs*j|ddlmfddS)Nr)residual_evaluator_ofcsjjjdS)N)rrrr)rrrr)datafilerrrrtrurs z4DataScan._build_residual_evaluator..)rrpyiceberg.expressions.visitorsr)rrrtrru_build_residual_evaluators z"DataScan._build_residual_evaluatorrrr$rcCs&|jtjkp|jtjko|jpt|kS)asEnsure that no manifests are loaded that contain deletes that are older than the data. Args: min_sequence_number (int): The minimal sequence number. manifest (ManifestFile): A ManifestFile that can be either data or deletes. Returns: Boolean indicating if it is either a data file, or a relevant delete file. )rir"rDELETESsequence_numberr1)rrrtrtru_check_sequence_numbers  zDataScan._check_sequence_numberIterator[list[ManifestEntry]]cst}|s tgStjfdd|jD}tjt|t }| ddfdd|DS)zFilter and return manifest entries based on partition and metrics evaluators. Returns: Iterator of ManifestEntry objects that match the scan's partition filter. csg|] }|j|r|qSrt)partition_spec_id)r manifest_file)manifest_evaluatorsrtrurs  z-DataScan.scan_plan_helper..cSst|Sr)r)argsrtrtrursz+DataScan.scan_plan_helper..cs0g|]}|rj||jfqSrt)rrrrr)rpartition_evaluatorsrrtrurs ) rziterr\rrrrrra get_or_createmap)rrzrexecutorrt)rrrrruscan_plan_helpers     zDataScan.scan_plan_helpercCs|jsdS|jS)z@Check if server-side scan planning should be used for this scan.F)rsupports_server_side_planningrrtrtru _should_use_server_side_plannings z)DataScan._should_use_server_side_planningIterable[FileScanTask]cCsddlm}ddlm}t|j|std|jdurtd||j |j dkr,t |j nd|j t kr5|j nd|jd}|j|j|S) z0Plan files using REST server-side scan planning.r) RestCatalog)PlanTableScanRequestz)REST scan planning requires a RestCatalogNz.REST scan planning requires a table identifierr)rr}r^r)pyiceberg.catalog.restrrrr1r TypeErrorrrrrr5rrr plan_scan)rrrrequestrtrtru_plan_files_server_side!s    z DataScan._plan_files_server_sidecsg}tt|jt|D]4}|j}|jtj kr"| |q|jtj kr1j ||j dq|jtjkr;tdtd|jd|fdd|DS)z(Plan files locally by reading manifests. partition_keyz^PyIceberg does not yet support equality deletes: https://github.com/apache/iceberg/issues/6568zUnknown DataFileContent (z): c sLg|]"}t|jj|jp t|j|jjd|jj|j|jjdqS)r)rr)rr< for_data_filerr1r>r residual_for)r data_entry delete_indexresidual_evaluatorsrtrurEs z.DataScan._plan_files_local..)r*r\rr from_iterablerr<rir!rr=POSITION_DELETESadd_delete_filer>EQUALITY_DELETESr)r data_entriesrr<rtrru_plan_files_local4s       zDataScan._plan_files_localcCs|r|S|S)aSPlans the relevant files by filtering on the PartitionSpecs. If the table comes from a REST catalog with scan planning enabled, this will use server-side scan planning. Otherwise, it falls back to local planning. Returns: List of FileScanTasks that contain both data and delete files. )rrrrrtrtrurZTs zDataScan.plan_filesr"cCs6ddlm}||j|j||j|j|j| S)zRead an Arrow table eagerly from this DataScan. All rows will be loaded into memory at once. Returns: pa.Table: Materialized Arrow Table from the Iceberg table's DataScan rrL) r0rLrrrrrrr]rZ)rrLrtrtrurbs  zDataScan.to_arrowpa.RecordBatchReadercCsbddl}ddlm}m}||}||j|j||j|j|j  | }|j |||S)aReturn an Arrow RecordBatchReader from this DataScan. For large results, using a RecordBatchReader requires less memory than loading an Arrow Table for the same DataScan, because a RecordBatch is read one at a time. Returns: pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg table's DataScan which can be used to read a stream of record batches one by one. rN)rLschema_to_pyarrow)r.r0rLrrrrrrrto_record_batchesrZRecordBatchReaderr|cast)rr8rLr target_schemabatchesrtrtrur{ps   zDataScan.to_arrow_batch_readerrr rcKs|jdi|S)zRead a Pandas DataFrame eagerly from this Iceberg table. Returns: pd.DataFrame: Materialized Pandas Dataframe from the Iceberg table Nrt)rrrrtrtrurr zDataScan.to_pandasNr9r connectionDuckDBPyConnection | NonerdcCs,ddl}|p |jdd}||||S)zShorthand for loading the Iceberg Table in DuckDB. Returns: DuckDBPyConnection: In memory DuckDB connection with the Iceberg table. rNz:memory:)database)duckdbconnectregisterr)rr9r rconrtrtru to_duckdbszDataScan.to_duckdbray.data.dataset.DatasetcCsddl}|j|S)zRead a Ray Dataset eagerly from this Iceberg table. Returns: ray.data.dataset.Dataset: Materialized Ray Dataset from the Iceberg table rN)raydata from_arrowr)rrrtrtruto_rayszDataScan.to_rayrcCs.ddl}||}t||jr|}|S)zRead a Polars DataFrame from this Iceberg table. Returns: pl.DataFrame: Materialized Polars Dataframe from the Iceberg table rN)rMrrr1Seriesto_frame)rrOresultrtrtrurPs  zDataScan.to_polarscCsddlm}d}|}|D]1}|jtkr$t|jdkr$||jj7}q||j |j | |j |j d}||g}|t|7}q|S)Nrr)rrrPrr)r0rLrZrrr_rr`rrrrrrr])rrLresrQtask arrow_scantblrtrtrur\s  zDataScan.count)rrlrr)rr)rrlrr)rrlrr)rr)rrlrr)rrlrr$rr)rr)rr)rrr)rrrr)r9rr r rrd)rrrrV)rorprqrr rrrrrrYrrrrrrZrr{rrrrPr\rtrtrtrurs,        ,       r)frozenc@sTeZdZUdZded<ded<ded<ded <d Zd ed <d Zd ed<dddZd S) WriteTaskz0Task with the parameters for writing a DataFile.z uuid.UUIDr-rltask_idr)rzlist[pa.RecordBatch]record_batchesNrrzPartitionKey | Noner extensionrrcCsd|jd|jd|S)Nz00000--.)r"r-)rr$rtrtrugenerate_data_file_filenamesz%WriteTask.generate_data_file_filename)r$rrr)rorprqrrrsrrr'rtrtrtrur!s   r!rr2rrIterable[DataFile]cs:ddlmtfdd|D}dd|DS)zConvert a list files into DataFiles. Returns: An iterable that supplies DataFiles that describe the parquet files. r)parquet_file_to_data_filecsg|] }|qSrt)submit)rrrrr)rrtrurz0_parquet_files_to_data_files..cSsg|] }|r|qSrt)r)rfrtrtrurr,)r0r)rar)rrrfuturesrtr+rurs r)rrrr)rrhrr ) rrrr$rrrrrr)rrrrl)rr2rrrrrr() __future__rr[r^rBrTabcrrcollections.abcrrr dataclassesr functoolsr r typesr typingr r rpydanticrpyiceberg.expressions.parser expressionsrrpyiceberg.expressionsrrrrrrrrrrrrrrr pyiceberg.iorrpyiceberg.manifestr r!r"r#r$pyiceberg.partitioningr%r&r'r(pyiceberg.schemar)!pyiceberg.table.delete_file_indexr*pyiceberg.table.inspectr,pyiceberg.table.locationsr-r.pyiceberg.table.maintenancer0pyiceberg.table.metadatar1r2pyiceberg.table.name_mappingr3pyiceberg.table.refsr4r5pyiceberg.table.snapshotsr6r7pyiceberg.table.sortingr8r9pyiceberg.table.updater:r;r<r=r>r?r@rArBrCrDrErFrGrHrIrJrKrLpyiceberg.table.update.schemarMpyiceberg.table.update.snapshotrNrOrPpyiceberg.table.update.sortingrRpyiceberg.table.update.specrT!pyiceberg.table.update.statisticsrVpyiceberg.transformsrWpyiceberg.typedefrXrYrZr[r\r]r^r_pyiceberg.typesr`pyiceberg.utils.concurrentrapyiceberg.utils.configrbpyiceberg.utils.propertiesrcrGrHrJrBpdrMrOr.r8rrrdrRrfpyiceberg.catalogrgrrhrirjrr3rkrvrrr5rrrrrrrZrrVrtrurrrrrrr!rrtrtrtrus      (      T     (       Vn: 9.  k 5 !  -