o Z>m?Z?m@Z@mAZAmBZBddlCmDZDddlEmFZFmGZGmHZHmIZIddlEmJZKddlLmMZMmNZNmOZOmPZPmQZQmRZRmSZSmTZTmUZUmVZVmWZWmXZXmYZYmZZZm[Z[m\Z\m]Z]m^Z^m_Z_m`Z`maZambZbmcZcmdZdmeZemfZfmgZgmhZhmiZimjZjmkZkmlZlmmZmmnZnmoZompZpmqZqmrZrmsZsmtZtmuZumvZvmwZwddlxmyZymzZzm{Z{ddl|m}Z}m~Z~mZmZmZddlmZmZmZmZmZmZmZmZmZmZmZmJZJmZddlmZmZddlmZddlmZddlmZmZddlmZddlmZmZddlmZmZmZmZdd lmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZdd!lmZdd"lmZdd#lmZdd$lmZdd%lmZmZmZdd&lmZdd'lmZmZerdd(lmZmZeeСZd)Zd*Zd+Zd,Zd-Zd.Zd/Zd0Zd1Zd2Zd3Zhd4Ze d5Zed/d:d;Zߐd0d>d?ZGd@dAdAeZGdBdCdCe$jjZGdDdEdEetevZGdFdGdGesZedHe{jfd1dRdSZGdTdUdUee%jZd2dZd[ZGd\d]d]eFe(jZGd^d_d_eFdZd3d4dddeZd3d4dfdgZed5dkdlZd6drdsZd7dzd{Zdd|ejfd8ddZd|ejfd9ddZd:ddZd:ddZed;ddZee%jdddZee%jd?ddZee%jd@ddZee%jdAddZee%jdBddZGdddeeeZdCddZGdddeeZGdddeeeBZGdddeee%jBZGdddeee%jBZGdddeZdDddZ ddejdfdEddÄZ dFddȄZ GddʄdʃZ d|d|ed|fdGdd҄Z GddԄdee%je%jdBfZGddքdee%jZdHddلZGddۄdeeZeZGdd݄d݃ZdސZdߐZGdddeZedHdGdddeɃZdIddZedHdGdddZGdddeeeZdJddZedHdGdddZGdddeeeZ dKddZ!edHdGdddZ"dLddZ#dMddZ$dNd d Z%d|ejfdOddZ&dPddZ'dQddZ(dZ)dZ*dRddZ+  dSdTd"d#Z,edHdGd$d%d%Z-dUd*d+Z.dVd-d.Z/dS(WauFileIO implementation for reading and writing table files that uses pyarrow.fs. This file contains a FileIO implementation that relies on the filesystem interface provided by PyArrow. It relies on PyArrow's `from_uri` method that infers the correct filesystem type to use. Theoretically, this allows the supported storage types to grow naturally with the pyarrow library. ) annotationsN)ABCabstractmethod)CallableIterableIterator)copy) dataclass)Enum) lru_cachesingledispatch) TYPE_CHECKINGAnyGenericTypeVarcast)urlparse) ChunkedArray)S3RetryStrategy)FileInfo FileSystemFileType)to_bytes) ResolveError) AlwaysTrueBooleanExpression BoundIsNaN BoundIsNull BoundTermNotOr)Literal)BoundBooleanExpressionVisitorbindextract_field_idstranslate_column_names)visit)+ADLS_ACCOUNT_KEYADLS_ACCOUNT_NAMEADLS_BLOB_STORAGE_AUTHORITYADLS_BLOB_STORAGE_SCHEMEADLS_CLIENT_IDADLS_CLIENT_SECRETADLS_DFS_STORAGE_AUTHORITYADLS_DFS_STORAGE_SCHEMEADLS_SAS_TOKENADLS_TENANT_IDAWS_ACCESS_KEY_ID AWS_REGION AWS_ROLE_ARNAWS_ROLE_SESSION_NAMEAWS_SECRET_ACCESS_KEYAWS_SESSION_TOKENGCS_DEFAULT_LOCATIONGCS_SERVICE_HOST GCS_TOKENGCS_TOKEN_EXPIRES_AT_MS HDFS_HOSTHDFS_KERB_TICKET HDFS_PORT HDFS_USERS3_ACCESS_KEY_ID S3_ANONYMOUSS3_CONNECT_TIMEOUT S3_ENDPOINTS3_FORCE_VIRTUAL_ADDRESSING S3_PROXY_URI S3_REGIONS3_REQUEST_TIMEOUTS3_RESOLVE_REGIONS3_RETRY_STRATEGY_IMPL S3_ROLE_ARNS3_ROLE_SESSION_NAMES3_SECRET_ACCESS_KEYS3_SESSION_TOKENFileIO InputFile InputStream OutputFile OutputStream)DataFileDataFileContent FileFormat)PartitionFieldPartitionFieldValue PartitionKey PartitionSpecpartition_record_value) PartnerAccessorPreOrderSchemaVisitorSchemaSchemaVisitorPerPrimitiveTypeSchemaWithPartnerVisitor_check_schema_compatiblebuild_position_accessorspre_order_visitpromote prune_columnssanitize_column_namesr&visit_with_partner$DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITETableProperties)load_location_provider) TableMetadata) NameMappingapply_name_mapping) PuffinFile)IdentityTransformTruncateTransform) EMPTY_DICT PropertiesRecord TableVersion) BinaryType BooleanTypeDateType DecimalType DoubleType FixedType FloatType IcebergType IntegerTypeListTypeLongTypeMapType NestedField PrimitiveType StringType StructTypeTimestampNanoType TimestampTypeTimestamptzNanoTypeTimestamptzTypeTimeType UnknownTypeUUIDType strtobool)ExecutorFactory)Config)millis_to_datetime)unscaled_to_decimal)get_first_property_valueproperty_as_boolproperty_as_int) Singleton)"truncate_upper_bound_binary_string truncate_upper_bound_text_string) FileScanTask WriteTaskiz buffer-sizesiceberg.schemasPARQUET:field_ids iceberg.idsiceberg.requiredsdocelementkeyvaluedoc>+00:00Etc/UTCZUTCTbucketstrreturn str | Nonec CsBddlm}z||dWSttfy td|YdSw)Nr)resolve_s3_regionrz$Unable to resolve region for bucket ) pyarrow.fsrOSError TypeErrorloggerwarning)rrrQ/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/pyiceberg/io/pyarrow.py_cached_resolve_s3_regions  rimplS3RetryStrategy | Nonec Csz.|d}t|dkrtd|d|dd|d}}t|}t||}|WSttfyDt j d|ddYdSw)N.zMretry-strategy-impl should be full path (module.CustomS3RetryStrategy), got: z(Could not initialize S3 retry strategy:  stacklevel) splitlen ValueErrorjoin importlib import_modulegetattrModuleNotFoundErrorAttributeErrorwarningswarn)r path_parts module_name class_namemoduleclass_rrr_import_retry_strategys    rcs"eZdZdZdfdd ZZS) UnsupportedPyArrowTypeExceptionz:Cannot convert PyArrow type to corresponding Iceberg type.fieldpa.Fieldargsrcs||_tj|dSN)rsuper__init__)selfrr __class__rrrsz(UnsupportedPyArrowTypeException.__init__)rrrr)__name__ __module__ __qualname____doc__r __classcell__rrrrrsrcseZdZd fdd ZZS) PyArrowLocalFileSystempathrrrkwargsrpyarrow.NativeFilecs0|jtj|ddtj|g|Ri|S)NT) recursive) create_dirosrdirnameropen_output_stream)rrrrrrrrsz)PyArrowLocalFileSystem.open_output_stream)rrrrrrrr)rrrrrrrrrrsrcseZdZUdZded<ded<ded<efd#fd d Zd$ddZd%ddZd&ddZ d'd(ddZ d)d*dd Z d+d!d"Z Z S), PyArrowFileaAA combined InputFile and OutputFile implementation using pyarrow filesystem. This class generates pyarrow.lib.NativeFile instances. Args: location (str): A URI or a path to a local file. Attributes: location(str): The URI or path to a local file for a PyArrowFile instance. Examples: >>> from pyiceberg.io.pyarrow import PyArrowFile >>> # input_file = PyArrowFile("s3://foo/bar.txt") >>> # Read the contents of the PyArrowFile instance >>> # Make sure that you have permissions to read/write >>> # file_content = input_file.open().read() >>> # output_file = PyArrowFile("s3://baz/qux.txt") >>> # Write bytes to a file >>> # Make sure that you have permissions to read/write >>> # output_file.create().write(b'foobytes') r _filesystemr_pathint _buffer_sizelocationrfs buffer_sizecs$||_||_||_tj|ddS)N)r)rrrrr)rrrrrrrrrszPyArrowFile.__init__rrc Csvz |j|j}Wn!ty*}z|jdksdt|vr%td|j|d}~ww|jt j kr9t d|j|S)zRetrieve a pyarrow.fs.FileInfo object for the location. Raises: PermissionError: If the file at self.location cannot be accessed due to a permission error such as an AWS error code 15. AWS Error [code 15]z%Cannot get file info, access denied: Nz&Cannot get file info, file not found: ) r get_file_inforrerrnorPermissionErrorrtyperNotFoundFileNotFoundError)r file_infoerrr _file_infos zPyArrowFile._file_infocCs|}|jS)z.Return the total length of the file, in bytes.)rsize)rrrrr__len__,szPyArrowFile.__len__boolcCs$z|WdStyYdSw)z"Check whether the location exists.TF)rrrrrrexists1s  zPyArrowFile.existsTseekablerOc Csz|r |j|j}W|S|jj|j|jd}W|Sttfy#tyW}z)|jdks5dt |vr>td|j ||jdksIdt |vrRtd|j |d}~ww) a!Open the location using a PyArrow FileSystem inferred from the location. Args: seekable: If the stream should support seek, or if it is consumed sequential. Returns: pyarrow.lib.NativeFile: A NativeFile instance for the file located at `self.location`. Raises: FileNotFoundError: If the file at self.location does not exist. PermissionError: If the file at self.location cannot be accessed due to a permission error such as an AWS error code 15. rrPath does not existz"Cannot open file, does not exist: rrz!Cannot open file, access denied: N) ropen_input_fileropen_input_streamrrrrrrr)rr input_filerrrropen9s   zPyArrowFile.openF overwriterQc Csz|s|durtd|j|jj|j|jd}W|Sty%tyE}z|j dks7dt |vr@td|j|d}~ww)aCreate a writable pyarrow.lib.NativeFile for this PyArrowFile's location. Args: overwrite (bool): Whether to overwrite the file if it already exists. Returns: pyarrow.lib.NativeFile: A NativeFile instance for the file located at self.location. Raises: FileExistsError: If the file already exists at `self.location` and `overwrite` is False. Note: This retrieves a pyarrow NativeFile by opening an output stream. If overwrite is set to False, a check is first performed to verify that the file does not exist. This is not thread-safe and a possibility does exist that the file can be created by a concurrent process after the existence check yet before the output stream is created. In such a case, the default pyarrow behavior will truncate the contents of the existing file when opening the output stream. Tz$Cannot create file, already exists: rrrz#Cannot create file, access denied: N) rFileExistsErrorrrrrrrrrr)rr output_filerrrrcreateVs zPyArrowFile.createcCs|S)arReturn a new PyArrowFile for the location of an existing PyArrowFile instance. This method is included to abide by the OutputFile abstract base class. Since this implementation uses a single PyArrowFile class (as opposed to separate InputFile and OutputFile implementations), this method effectively returns a copy of the same instance. rrrrr to_input_fileuszPyArrowFile.to_input_file)rrrrrrrr)rr)rr)rr)T)rrrrO)F)rrrrQ)rr)rrrr__annotations__ ONE_MEGABYTErrrrrrrrrrrrrs      rcseZdZUded<efd/fdd Zeefd0d d Zd1d2ddZd3ddZ d4ddZ d3ddZ d2ddZ d3ddZ d3ddZd5d!d"Zd5d#d$Zd6d'd(Zd7d*d+Zd8d-d.ZZS)9 PyArrowFileIOz'Callable[[str, str | None], FileSystem] fs_by_scheme propertiesrqcst|j|_tj|ddS)N)r)r _initialize_fsrrr)rrrrrrs zPyArrowFileIO.__init__rrrtuple[str, str, str]cCslt|}|js|dd}|dd}||tj|fS|jdvr)|j|j|jfS|j|j|j|jfS)zReturn (scheme, netloc, path) for the given location. Uses DEFAULT_SCHEME and DEFAULT_NETLOC if scheme/netloc are missing. DEFAULT_SCHEMEfileDEFAULT_NETLOC)hdfsviewfs)rschemegetrrabspathnetloc)rruridefault_schemedefault_netlocrrrparse_locations   zPyArrowFileIO.parse_locationNrrrrcCst|dvr|S|dvr||S|dvr|||S|dvr#|S|dvr+|S|dvr3|Std|)z+Initialize FileSystem for different scheme.>oss>s3s3as3n>rr>gsgcs>abfswasbabfsswasbs>r z%Unrecognized filesystem type in URI: )_initialize_oss_fs_initialize_s3_fs_initialize_hdfs_fs_initialize_gcs_fs_initialize_azure_fs_initialize_local_fsr)rrrrrrrs  zPyArrowFileIO._initialize_fsc Csddlm}|jtt|jttt|jtt t|jt t t|jt t t|jtdd}|jt}r8||d<|jt}rFt||d<|jt}rTt||d<t|jtt}ra||d<t|jtt}rn||d <|jt}r|t||d <|d i|S) Nr S3FileSystemT)endpoint_override access_key secret_key session_tokenregionforce_virtual_addressing proxy_optionsconnect_timeoutrequest_timeoutrole_arn session_name anonymousr)rr)rrrBrr?r1rKr5rLr6rEr2rrCrDrAfloatrFrIr3rJr4r@r) rr) client_kwargs proxy_urir1r2r3r4 s3_anonymousrrrr"s*           z PyArrowFileIO._initialize_oss_fscCsddlm}t|jtt}|dust|jtddur8t|dp |}|dur7||kr7t d|d|d|n|}|j t t|jt tt|jttt|jtt|d }|j t}ra||d <|j t}rot||d <|j t}r}t||d <t|jtt} r| |d <t|jtt} r| |d<|j tdurt|jtd|d<|j t} rt| } r| |d<|j t} rt| |d<|di|S)Nrr(FTrz6PyArrow FileIO overriding S3 bucket region for bucket z: provided region z, actual region )r*r+r,r-r.r0r1r2r3r4r/retry_strategyr5r) rr)rrrEr2rrGrrrrrBr?r1rKr5rLr6rDrAr6rFrIr3rJr4rCrHrr@r)rrr)provided_region bucket_regionr7r8r1r2r3r4retry_strategy_implretry_instancer9rrrr#sN         zPyArrowFileIO._initialize_s3_fscsddlm}d}|tj||krtd|dtjdddlm}i|j t }r3|d<|j t }r?|d <|j t }rK|d <|j t }rW|d <|j t}rc|d <|j t} ro| d <|j t} r{| d<|j t} r| d<|j t} r| d<|j t} r| d<gd}fdd|D}|rt|t|krɇfdd|D}td|d||diS)Nr)versionz20.0.0zpyarrow version >= z9 required for AzureFileSystem support, but found version r)AzureFileSystem account_name account_keyblob_storage_authoritydfs_storage_authorityblob_storage_schemedfs_storage_scheme sas_token client_id client_secret tenant_id)rHrIrJcsg|]}|vr|qSrr.0rr7rr 1z6PyArrowFileIO._initialize_azure_fs..csg|]}|vr|qSrrrKrMrrrN3rOzclient_id, client_secret, and tenant_id must all be provided together to use ClientSecretCredential for Azure authentication. Provided: z , Missing: r) packagingr?parsepyarrow __version__ ImportErrorrr@rrr(r'r)r-r*r.r/r+r,r0rr)rr?'MIN_PYARROW_VERSION_SUPPORTING_AZURE_FSr@rArBrCrDrErFrGrHrIrJcredential_keys provided_keys missing_keysrrMrr&sT  z"PyArrowFileIO._initialize_azure_fsc Csddlm}i}|r||d|S|jt}r ||d<|jt}r.t||d<|jt}r:||d<|jt }rF||d<|di|S) Nr)HadoopFileSystemz://hostportuser kerb_ticketr) rrYfrom_urirrr;r=rr>r<) rrrrY hdfs_kwargsrZr[r\r]rrrr$<s  z!PyArrowFileIO._initialize_hdfs_fscCsddlm}i}|jt}r||d<|jt}r$tt||d<|jt}r0||d<|jt }rFt |}|j |d<|j |d<|di|S) Nr) GcsFileSystem access_tokencredential_token_expirationdefault_bucket_locationrr*r) rr`rrr9r:rrr7r8rrr)rr` gcs_kwargsra expirationbucket_locationendpoint url_partsrrrr%Ns   z PyArrowFileIO._initialize_gcs_fscCstSr)rrrrrr'_z"PyArrowFileIO._initialize_local_fsrc C:|||j\}}}t|||||t|jttdS)zGet a PyArrowFile instance to read bytes from the file at the given location. Args: location (str): A URI or a path to a local file. Returns: PyArrowFile: A PyArrowFile instance for the given location. rrrrrrrrrr BUFFER_SIZErrrrrrrrr new_inputb  zPyArrowFileIO.new_inputc Crj)zGet a PyArrowFile instance to write bytes to the file at the given location. Args: location (str): A URI or a path to a local file. Returns: PyArrowFile: A PyArrowFile instance for the given location. rkrlrnrrr new_outputsrpzPyArrowFileIO.new_outputstr | InputFile | OutputFileNonec Cst|ttfr |jn|}|||j\}}}|||}z||WdSty,t y3t ye}z'|j dksEdt |vrMtd|||j dksXdt |vr`t d||d}~ww)a=Delete the file at the given location. Args: location (Union[str, InputFile, OutputFile]): The URI to the file--if an InputFile instance or an OutputFile instance is provided, the location attribute for that instance is used as the location to delete. Raises: FileNotFoundError: When the file at the provided location does not exist. PermissionError: If the file at the provided location cannot be accessed due to a permission error such as an AWS error code 15. rrz$Cannot delete file, does not exist: rrz#Cannot delete file, access denied: N) isinstancerNrPrrrr delete_filerrrrr)rr str_locationrrrrrrrrdeletes"    zPyArrowFileIO.deletedict[str, Any]cCst|j}d|d<|S)zCCreate a dictionary of the PyArrowFileIO fields used when pickling.Nr)r__dict__)r fileio_copyrrr __getstate__s zPyArrowFileIO.__getstate__statecCs||_t|j|_dS)z4Deserialize the state into a PyArrowFileIO instance.N)ryr rr)rr|rrr __setstate__szPyArrowFileIO.__setstate__)rrq)rrrrqrr r)rrrrrr)rr)rrrr)rrrr)rrrrrsrrx)r|rxrrs)rrrrrpr staticmethodrrr"r#r&r$r%r'rorqrwr{r}rrrrrrs"    6 8      rTschemaSchema | IcebergTypemetadatadict[bytes, bytes]include_field_idsr file_formatrT pa.schemacCst|t|||Sr)r&_ConvertToArrowSchema)rrrrrrrschema_to_pyarrowsrc@seZdZUded<eddfdad d ZdbddZdcddZdddd Zded$d%Z dfd*d+Z dgd.d/Z dhd2d3Z did5d6Z djd8d9Zdkd;d<Zdld>d?ZdmdAdBZdndDdEZdodGdHZdpdJdKZdqdMdNZdrdPdQZdsdSdTZdtdVdWZdudYdZZdvd\d]Zdwd_d`ZdS)xrr _metadataTNrrrrFileFormat | NonerrscCs||_||_||_dSr)r_include_field_ids _file_format)rrrrrrrr z_ConvertToArrowSchema.__init___r\ struct_result pa.StructTypercCstjt||jdS)N)r)parlistr)rrrrrrrz_ConvertToArrowSchema.schemar field_resultsbuiltins.list[pa.DataType] pa.DataTypecC t|Srrstruct)rrrrrrr z_ConvertToArrowSchema.structrr field_resultrcCsxi}|jr |j|t<|jr"|jtjkrt|j|t<nt|j|t <|jtjkr1t|j  |t <t j|j||j|dSN)namernullabler)rPYARROW_FIELD_DOC_KEYrrrTORCrfield_idORC_FIELD_ID_KEYPYARROW_PARQUET_FIELD_ID_KEYrequiredlowerORC_FIELD_REQUIRED_KEYrrroptional)rrrrrrrrs   z_ConvertToArrowSchema.field list_typer}element_resultcCs||j|}tj|dS)N) value_type)r element_fieldr large_list)rrrrrrrrs z_ConvertToArrowSchema.listmap_typer key_result value_resultcCs*||j|}||j|}tj||dS)N)key_type item_type)r key_field value_fieldrmap_)rrrrrrrrrmapsz_ConvertToArrowSchema.map fixed_typerycCstt|Sr)rbinaryrrrrrr visit_fixedz!_ConvertToArrowSchema.visit_fixed decimal_typerwcCst|j|jSr)r decimal128 precisionscalerrrrr visit_decimalsz#_ConvertToArrowSchema.visit_decimalrucCtSr)rbool_rrrrr visit_booleanz#_ConvertToArrowSchema.visit_booleanr|cCrr)rint32rrrr visit_integerrz#_ConvertToArrowSchema.visit_integerr~cCrr)rint64rrrr visit_longrz _ConvertToArrowSchema.visit_longrzcCrr)rfloat32rrrr visit_floatz!_ConvertToArrowSchema.visit_floatrxcCrr)rfloat64rrrr visit_doublerz"_ConvertToArrowSchema.visit_doublervcCrr)rdate32rrrr visit_daterz _ConvertToArrowSchema.visit_datercC tdS)Nus)rtime64rrrr visit_timerz _ConvertToArrowSchema.visit_timercC tjddS)Nrunitr timestamprrrrvisit_timestamp  z%_ConvertToArrowSchema.visit_timestamprcCr)Nnsrrrrrrvisit_timestamp_ns rz(_ConvertToArrowSchema.visit_timestamp_nsrcCtjdddS)Nrrrtzrrrrrvisit_timestamptzrz'_ConvertToArrowSchema.visit_timestamptzrcCr)Nrrrrrrrrvisit_timestamptz_nsrz*_ConvertToArrowSchema.visit_timestamptz_nsrcCrr)r large_stringrrrr visit_stringrz"_ConvertToArrowSchema.visit_stringrcCrr)ruuidrrrr visit_uuidrz _ConvertToArrowSchema.visit_uuidrcCr)z\Type `UnknownType` can be promoted to any primitive type in V3+ tables per the Iceberg spec.)rnullrrrr visit_unknownrz#_ConvertToArrowSchema.visit_unknownrtcCrr)r large_binaryrrrr visit_binaryrz"_ConvertToArrowSchema.visit_binary)rrrrrrrrs)rr\rrrr)rrrrrr)rrrrrr)rr}rrrr)rrrrrrrr)rryrr)rrwrr)rrurr)rr|rr)rr~rr)rrzrr)rrxrr)rrvrr)rrrr)rrrr)rrrr)rrrr)rrrr)rrrr)rrrr)rrrr)rrtrr)rrrrrprrrrrrrrrrrrrrrrrrrrrrrrrrrrs4                     rr iceberg_typer{ pa.scalarcCs*t|ts td|tj|t|dS)NExpected primitive type, got: )rr)rtrrrscalarr)rrrrr_convert_scalar#s rc@seZdZUdZded<dd d Zd?ddZd?ddZd@ddZ d@ddZ d@ddZ d@ddZ dAdd Z dAd!d"ZdAd#d$ZdAd%d&ZdAd'd(ZdAd)d*ZdAd+d,ZdAd-d.ZdBd/d0ZdBd1d2ZdCd4d5ZdDd8d9ZdDd:d;ZdS)E_ConvertToArrowExpressionzConvert Iceberg bound expressions to PyArrow expressions. Args: schema: Optional Iceberg schema to resolve full field paths for nested fields. If not provided, only the field name will be used (not dotted path). Schema | None_schemaNrcC ||_dSr)r)rrrrrr3rz"_ConvertToArrowExpression.__init__termrrstr | tuple[str, ...]cCsL|jdur |j|jj}|dur d|vrt|dS|S|jjS)a@Get the field name or nested field path for a bound term. For nested struct fields, returns a tuple of field names (e.g., ("mazeMetadata", "run_id")). For top-level fields, returns just the field name as a string. PyArrow requires nested field references as tuples, not dotted strings. Nr)rfind_column_namerefrrtuplerr)rr full_namerrr_get_field_name6s  z)_ConvertToArrowExpression._get_field_nameliteralsset[Any] pc.ExpressioncCs0tj|t|jjd}t|||SNr rarrayrrr field_typepcrisinrrrpyarrow_literalsrrrvisit_inJsz"_ConvertToArrowExpression.visit_incCs2tj|t|jjd}t|||Srrrrrr visit_not_inNsz&_ConvertToArrowExpression.visit_not_incCst||}t|Srrrris_nanrrrrrr visit_is_nanRs z&_ConvertToArrowExpression.visit_is_nancCst||}t|Srr r rrr visit_not_nanVs z'_ConvertToArrowExpression.visit_not_nancCst||jddS)NF) nan_is_null)rrris_nullrrrrr visit_is_nullZz'_ConvertToArrowExpression.visit_is_nullcCst||Sr)rrris_validrrrrvisit_not_null]rz(_ConvertToArrowExpression.visit_not_nullliteral Literal[Any]cCs$t||t|j|jjkSrrrrrrrrrrrrrr visit_equal`$z%_ConvertToArrowExpression.visit_equalcCs$t||t|j|jjkSrrrrrrvisit_not_equalcrz)_ConvertToArrowExpression.visit_not_equalcCs$t||t|j|jjkSrrrrrrvisit_greater_than_or_equalfrz5_ConvertToArrowExpression.visit_greater_than_or_equalcCs$t||t|j|jjkSrrrrrrvisit_greater_thanirz,_ConvertToArrowExpression.visit_greater_thancCs$t||t|j|jjkSrrrrrrvisit_less_thanlrz)_ConvertToArrowExpression.visit_less_thancCs$t||t|j|jjkSrrrrrrvisit_less_than_or_equalorz2_ConvertToArrowExpression.visit_less_than_or_equalcCstt|||jSrr starts_withrrrrrrrvisit_starts_withrz+_ConvertToArrowExpression.visit_starts_withcCstt|||jSrr rrrrvisit_not_starts_withusz/_ConvertToArrowExpression.visit_not_starts_withcCrNTrrrrrr visit_truexrz$_ConvertToArrowExpression.visit_truecCr)NFr&rrrr visit_false{rz%_ConvertToArrowExpression.visit_false child_resultcCs|Srrrr)rrr visit_not~riz#_ConvertToArrowExpression.visit_not left_result right_resultcCs||@Srrrr,r-rrr visit_andrz#_ConvertToArrowExpression.visit_andcCs||BSrrr.rrrvisit_orrz"_ConvertToArrowExpression.visit_orr)rr)rrrr)rrrrrr)rrrr)rrrrrr)rr)r)rrr)r,rr-rrr)rrrrrrrrrr r rrrrrrrrr"r$r'r(r+r/r0rrrrr)s0                     rcs0eZdZUded<ded<ded<ded<dEfdd ZdFd d ZdFddZdFddZdFddZdGddZ dGddZ dFddZ dFddZ dFddZ dFd d!ZdHd$d%ZdHd&d'ZdHd(d)ZdHd*d+ZdHd,d-ZdHd.d/ZdHd0d1ZdHd2d3ZdEd4d5ZdEd6d7ZdId9d:ZdJd=d>ZdJd?d@ZdKdCdDZZS)L!_NullNaNUnmentionedTermsCollectorzset[BoundTerm]is_null_or_not_bound_termsnull_unmentioned_bound_termsis_nan_or_not_bound_termsnan_unmentioned_bound_termsrrscs.tt|_t|_t|_t|_dSr)rrsetr2r3r4r5rrrrrs  z*_NullNaNUnmentionedTermsCollector.__init__rrcC&||jvr |j||j|dS)zJHandle the predicate case where either is_null or is_not_null is included.N)r3remover2addrrrr_handle_explicit_is_null_or_not  zA_NullNaNUnmentionedTermsCollector._handle_explicit_is_null_or_notcC||jvr |j|dSdS)zKHandle the predicate case where neither is_null or is_not_null is included.N)r2r3r9rrrr_handle_null_unmentioned z:_NullNaNUnmentionedTermsCollector._handle_null_unmentionedcCr7)zHHandle the predicate case where either is_nan or is_not_nan is included.N)r5r8r4r9rrrr_handle_explicit_is_nan_or_notr;z@_NullNaNUnmentionedTermsCollector._handle_explicit_is_nan_or_notcCr<)zIHandle the predicate case where neither is_nan or is_not_nan is included.N)r4r5r9rrrr_handle_nan_unmentionedr>z9_NullNaNUnmentionedTermsCollector._handle_nan_unmentionedrrcC||||dSrr=r@rrrrrrr z*_NullNaNUnmentionedTermsCollector.visit_incCrArrBrCrrrrrDz._NullNaNUnmentionedTermsCollector.visit_not_incCrArr=r?rrrrr rDz._NullNaNUnmentionedTermsCollector.visit_is_nancCrArrErrrrr rDz/_NullNaNUnmentionedTermsCollector.visit_not_nancCrArr:r@rrrrrrDz/_NullNaNUnmentionedTermsCollector.visit_is_nullcCrArrFrrrrrrDz0_NullNaNUnmentionedTermsCollector.visit_not_nullrrcCrArrBrrrrrrDz-_NullNaNUnmentionedTermsCollector.visit_equalcCrArrBrrrrrrDz1_NullNaNUnmentionedTermsCollector.visit_not_equalcCrArrBrrrrrrDz=_NullNaNUnmentionedTermsCollector.visit_greater_than_or_equalcCrArrBrrrrrrDz4_NullNaNUnmentionedTermsCollector.visit_greater_thancCrArrBrrrrrrDz1_NullNaNUnmentionedTermsCollector.visit_less_thancCrArrBrrrrrrDz:_NullNaNUnmentionedTermsCollector.visit_less_than_or_equalcCrArrBrrrrr"rDz3_NullNaNUnmentionedTermsCollector.visit_starts_withcCrArrBrrrrr$rDz7_NullNaNUnmentionedTermsCollector.visit_not_starts_withcCdSrrrrrrr'z,_NullNaNUnmentionedTermsCollector.visit_truecCrGrrrrrrr(rHz-_NullNaNUnmentionedTermsCollector.visit_falser)cCrGrrr*rrrr+rHz+_NullNaNUnmentionedTermsCollector.visit_notr,r-cCrGrrr.rrrr/rHz+_NullNaNUnmentionedTermsCollector.visit_andcCrGrrr.rrrr0rHz*_NullNaNUnmentionedTermsCollector.visit_orexprrcCst||dS)zCollect bound references categorized by null predicates. Categorizes by having at least one is_null or is_not_null in the expr and the remaining. N)boolean_expression_visit)rrIrrrcollectsz)_NullNaNUnmentionedTermsCollector.collectrrs)rrrrs)rrrrrrs)rrrrrrs)r)rsrrs)r,rsr-rsrrs)rIrrrs)rrrrrr:r=r?r@rrr r rrrrrrrrr"r$r'r(r+r/r0rKrrrrrr1s<                        r1rIrrrcCst|t|S)arConvert an Iceberg boolean expression to a PyArrow expression. Args: expr: The Iceberg boolean expression to convert. schema: Optional Iceberg schema to resolve full field paths for nested fields. If provided, nested struct fields will use dotted paths (e.g., "parent.child"). Returns: A PyArrow compute expression. )rJr)rIrrrrexpression_to_pyarrows rMcCszt}||t|jddd}t|jddd}t|}|D] }t|t|d}q |D] }t|t|d}q-t ||S)zComplementary filter conversion function of expression_to_pyarrow. Could not use expression_to_pyarrow(Not(expr)) to achieve this complementary effect because ~ in pyarrow.compute.Expression does not handle null. cS |jjSrrrrrrrr z6_expression_to_complementary_pyarrow..)rcSrNrrOrPrrrrQrRrP) r1rKsortedr3r5rr rrrM)rIr collectorr3r5 preserve_exprrrrr$_expression_to_complementary_pyarrows    rVrrx ds.FileFormatcKsT|tjkr tjdi|S|tjkr#dd|D}tjdi|Std|)NcSsi|] \}}|dvr||qS) pre_bufferrr)rLkvrrr .sz$_get_file_format..zUnsupported file format: r)rTPARQUETdsParquetFileFormatritems OrcFileFormatr)rr orc_kwargsrrr_get_file_format(s  rciorM data_filerRdict[str, pa.ChunkedArray]csl|jtjkrG||j}t|jddtd|}t j j |d Wdn1s.wY fdddjdjDS|jtjkr||j'}t|j|}t j j |d fd ddDWdS1swYdS|jtjkr||j }|}Wdn1swYt|Std |j) N) file_pathT)dictionary_columnsrYr)fragmentc,i|]}|td|kdqSrgposas_pyfilterrrcolumn)rLr tablerrr\<z!_read_deletes..rgrcrjrkrm)rLrrqrrr\Ersz"Delete file format not supported: )rrTr]rorgrrcr make_fragmentr^Scanner from_fragmentto_tableunify_dictionariesrpchunks dictionaryruniquePUFFINreadrm to_vectorr)rdrefidelete_fragmentpayloadrrqr _read_deletes4s6      $   rpositional_deleteslist[pa.ChunkedArray] start_indexr end_indexpa.Arrayc Csnt|dkr |d}n ttjdd|D}tt||}t|t tj ||d}t |t |S)NrcSg|]}|jqSr)ry)rLarrrrrrNVz/_combine_positional_deletes..) value_set) rr chunked_array itertoolschainrrangerroinvertis_insubtractr)rrr all_chunks full_rangeresultrrr_combine_positional_deletesRs  rF pa.Schema name_mappingNameMapping | Nonedowncast_ns_timestamp_to_usformat_versionrsr\cCsJt|t}|rt|t||dS|dur!t|||d}t||Std)NrrznParquet file does not have field-ids and the Iceberg table does not have 'schema.name-mapping.default' defined) visit_pyarrow_HasIds_ConvertToIceberg_pyarrow_to_schema_without_idsrlr)rrrrhas_idsschema_without_idsrrrpyarrow_to_schemads   rcCst|t||dS)Nr)r_ConvertToIcebergWithoutIDs)rrrrrrrzs rcC t|tSr)r_ConvertToLargeTypesrrrr"_pyarrow_schema_ensure_large_typesrrcCrr)r_ConvertToSmallTypesrrrr"_pyarrow_schema_ensure_small_typesrrobjpa.DataType | pa.SchemavisitorPyArrowSchemaVisitor[T]cCstd|)aApply a pyarrow schema visitor to any point within a schema. The function traverses the schema in post-order fashion. Args: obj (Union[pa.DataType, pa.Schema]): An instance of a Schema or an IcebergType. visitor (PyArrowSchemaVisitor[T]): An instance of an implementation of the generic PyarrowSchemaVisitor base class. Raises: NotImplementedError: If attempting to visit an unrecognized object type. zCannot visit non-type: )NotImplementedErrorrrrrrrs rcCs||tt||Sr)rrrrrrrrrsrrcsfdd|D}||S)Ncsg|]}t|qSr)rrLrrrrrNsz_..)r)rrresultsrrrrs 5pa.ListType | pa.LargeListType | pa.FixedSizeListTypecCs0||jt|j|}||j|||Sr)before_list_elementrrrafter_list_elementr)rrrrrrrs     pa.MapTypecCsV||jt|j|}||j||jt|j|}||j| |||Sr) before_map_keyrrr after_map_keybefore_map_value item_fieldrafter_map_valuer)rrrrrrrrs      pa.DictionaryTypecCs*tdt|d|jdt|j|S)Nz)Iceberg does not have a dictionary type. z will be inferred as z on read.)rrrrrrrrrrs rc Csh|j}||zt||}Wnty(}z t|d|jd||d}~ww|||||S)NzColumn 'z' has an unsupported type: )r before_fieldrrrr after_fieldr)rrrrrrrrrs   rcCs(tj|rtdt|||S)Nr)rtypes is_nestedrr primitiverrrrrs  c@seZdZd7ddZd7ddZd8d d Zd8d d Zd9ddZd9ddZd:ddZ d:ddZ e d;ddZ e dd*d+Ze d?d0d1Ze d@d4d5Zd6S)APyArrowSchemaVisitorrrrrscCdS)zNOverride this method to perform an action immediately before visiting a field.Nrrrrrrrz!PyArrowSchemaVisitor.before_fieldcCr)zMOverride this method to perform an action immediately after visiting a field.Nrrrrrrrz PyArrowSchemaVisitor.after_fieldrcCr)zcOverride this method to perform an action immediately before visiting an element within a ListType.Nrrrrrrrrz(PyArrowSchemaVisitor.before_list_elementcCr)zbOverride this method to perform an action immediately after visiting an element within a ListType.Nrrrrrrrz'PyArrowSchemaVisitor.after_list_elementrcCr)z]Override this method to perform an action immediately before visiting a key within a MapType.Nrrrrrrrrz#PyArrowSchemaVisitor.before_map_keycCr)z\Override this method to perform an action immediately after visiting a key within a MapType.Nrrrrrrrz"PyArrowSchemaVisitor.after_map_keyrcCr)z_Override this method to perform an action immediately before visiting a value within a MapType.Nrrrrrrrrz%PyArrowSchemaVisitor.before_map_valuecCr)z^Override this method to perform an action immediately after visiting a value within a MapType.Nrrrrrrrz$PyArrowSchemaVisitor.after_map_valuerrrrcCr)zVisit a schema.NrrrrrrrrrzPyArrowSchemaVisitor.schemarrrbuiltins.list[T]cCr)zVisit a struct.NrrrrrrrrrzPyArrowSchemaVisitor.structrcCr)zVisit a field.NrrrrrrrrrzPyArrowSchemaVisitor.fieldr pa.ListTypercCr)z Visit a list.NrrrrrrrrrzPyArrowSchemaVisitor.listrrrrcCr)z Visit a map.NrrrrrrrrrrzPyArrowSchemaVisitor.maprrcCr)zVisit a primitive type.Nrrrrrrr rzPyArrowSchemaVisitor.primitiveNrrrrsrrrrsrrrrsrrrrs)rrrrrr)rrrrrr)rrrrrr)rrrrrr)rrrrrrrr)rrrr)rrrrrrrrrrrrrrrrrrrrrrrs*             rr int | NonecCsB|jr|jt}rt|S|jt}rt|SdS)zFReturn the Iceberg field ID from Parquet or ORC metadata if available.N)rrrrdecoder)rfield_id_bytesrrr _get_field_ids   rc@sHeZdZd#ddZd$d d Zd%ddZd&ddZd'ddZd(d d!Zd"S))rrrrrrcC|Srrrrrrr rHz_HasIds.schemarrrbuiltins.list[bool]cCst|Sr)allrrrrr#rz_HasIds.structrrrcCstt|du|gSr)rrrrrrr&rz _HasIds.fieldrrrcCs|j}t|}|o |duSr)rrrrrr element_idrrrr)s z _HasIds.listrrrrcCs4|j}t|}|j}t|}t|du|du||gSr)rrrrrrrrrkey_idrvalue_idrrrr.s z _HasIds.maprrcCrr%rrrrrr5rHz_HasIds.primitiveN)rrrrrr)rrrrrr)rrrrrr)rrrrrr)rrrrrrrr)rrrr rrrrrrrrrrrrrrs     rc@seZdZUdZded<dejfdJd d ZdKddZdLddZ dMddZ dNd"d#Z dOd(d)Z dPd/d0Z dQd4d5ZdRd6d7ZdRd8d9ZdSd;d<ZdSd=d>ZdTd@dAZdSdBdCZdUdEdFZdSdGdHZdIS)VrzXConverts PyArrowSchema to Iceberg Schema. Applies the IDs from name_mapping if provided.builtins.list[str] _field_namesFrrrrsrrscCsg|_||_||_dSr)r_downcast_ns_timestamp_to_us_format_version)rrrrrrr>rz_ConvertToIceberg.__init__rrrcCs$t|}dur |Std|d)NzCannot convert z' to Iceberg Field as field_id is empty.)rr)rrrrrr _field_idEsz_ConvertToIceberg._field_idrrrrr\cCs t|jSr)r\fieldsrrrrrKrz_ConvertToIceberg.schemarrrbuiltins.list[NestedField]cCst|Sr)rrrrrrNrz_ConvertToIceberg.structrr{rcCsH||}|jr|jt}r|nd}|}t||j||j |dS)N)rr)rrrrrrrr)rrrrdoc_str field_docrrrrrQs "z_ConvertToIceberg.fieldrrrr}cCs8|j}|jt||}|jt|||j dS)N)element_required)rrappendLIST_ELEMENT_NAMErpopr}rrrrrrWs    z_ConvertToIceberg.listrrrrrcCsb|j}|jt||}|j|j}|jt||}|jt|||||j dS)N)value_required) rrr MAP_KEY_NAMErrrMAP_VALUE_NAMErrrrrrr^s      z_ConvertToIceberg.maprrrcCsHtj|r tStj|r'|j}|dkrtS|dkr tStd|tj |r0t Stj |r9t St |tjrLttj|}t|j|jStj|s^tj|s^tj|ratStj|rjtSt |tjrx|jdkrxtStj|rttj|}|jdvrn*|jdkr|jrt dn|j!dkr|j"t#vrt$St%Std td |j|j"t#vrt&S|j"durtSnXtj'|stj(|stj)|rt*Stj+|rttj,|}t-|j.Stj/|r|j!dkr|j0rd 1|j0nd }t2d |j!d|dt3St |tj4rt5Std|)N @zUnsupported integer type: r)smsrrzKIceberg does not yet support 'ns' timestamp precision. Downcasting to 'us'.zIceberg does not yet support 'ns' timestamp precision. Use 'downcast-ns-timestamp-to-us-on-write' configuration property to automatically downcast 'ns' to 'us' on write.z*Unsupported precision for timestamp type: rzzANull type (pa.null()) is not supported in Iceberg format version z . Field: zQ. Requires format-version=3+ or use a concrete type (string, int, boolean, etc.).zUnsupported type: )6rr is_booleanru is_integer bit_widthr|r~r is_float32rz is_float64rxrtDecimal128Typerrwrr is_stringis_large_stringis_string_viewr is_date32rv Time64Typerr is_timestamprrrrrr UTC_ALIASESrrr is_binaryis_large_binaryis_binary_viewrtis_fixed_size_binaryFixedSizeBinaryTypery byte_widthrrrrrUuidTyper)rrwidth field_pathrrrrisz      $          $    z_ConvertToIceberg.primitivecCs|j|jdSr)rrrrrrrrsz_ConvertToIceberg.before_fieldcC|jdSrrrrrrrrrz_ConvertToIceberg.after_fieldrcC|jtdSr)rrrrrrrrz%_ConvertToIceberg.before_list_elementcCrrrrrrrrrz$_ConvertToIceberg.after_list_elementrcCrr)rrrrrrrrrz _ConvertToIceberg.before_map_keycCrrrrrrrrrz_ConvertToIceberg.after_map_keyrcCrr)rrrrrrrrrz"_ConvertToIceberg.before_map_valuecCrrrrrrrrrz!_ConvertToIceberg.after_map_valueN)rrrrsrrsrrrr)rrrrrr\)rrrrrr)rrrr{rr)rrrr{rr})rrrr{rr{rr)rrrrrrrr)rrrrrrhDEFAULT_FORMAT_VERSIONrrrrrrrrrrrrrrrrrrrrr9s(        I      rc@HeZdZd"ddZd#d d Zd$ddZd%ddZd&ddZd'dd Zd!S)(rrrrrrcCrrrrrrrrrrz_ConvertToLargeTypes.schemarrbuiltins.list[pa.Field]cCrrrrrrrrrz_ConvertToLargeTypes.structrrrrcC ||Sr with_typerrrrrrz_ConvertToLargeTypes.fieldrrrcCrr)rrrrrrrrz_ConvertToLargeTypes.listrrrrcC t||Srrrrrrrrrz_ConvertToLargeTypes.maprcC,|tkr tS|tkrtS|Sr)rstringrrrrrrrr  z_ConvertToLargeTypes.primitiveNrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrrr     rc@r)(rrrrrrcCrrrrrrrrrz_ConvertToSmallTypes.schemarrrcCrrrrrrrrrz_ConvertToSmallTypes.structrrrrcCrrrrrrrrrz_ConvertToSmallTypes.fieldrrrcCrr)rlist_rrrrrrz_ConvertToSmallTypes.listrrrrcCrrrrrrrrrz_ConvertToSmallTypes.maprcCrr)rrrrrrrrrrrz_ConvertToSmallTypes.primitiveNr r!r"r#r$r%rrrrrrr&rc@seZdZdZd ddZdS) raa Converts PyArrowSchema to Iceberg Schema with all -1 ids. The schema generated through this visitor should always be used in conjunction with `new_table_metadata` function to assign new field ids in order. This is currently used only when creating an Iceberg Schema from a PyArrow schema when creating a new Iceberg table. rrrrcCr)NrrrrrrrrHz%_ConvertToIcebergWithoutIDs._field_idNr)rrrrrrrrrrs rr projected_schema table_schemapartition_specPartitionSpec | Nonefile_project_field_idsset[int]dict[int, Any]c Cs|j|}t|dks|durtS||}t|}i}|D]} || D]} t| jt r=|| j  |j } r=| || <q&q|S)z-Apply Column Projection rules to File Schema.rN) field_ids differencerrppartition_typer`fields_by_source_idrt transformrnrr partition) r r(r)r*r,project_schema_diffpartition_schema accessorsprojected_missing_fieldsrpartition_fieldpartition_valuerrr_get_column_projection_values s   r;taskrbound_row_filterprojected_field_idslist[ChunkedArray] | Nonecase_sensitive bool | NoneIterator[pa.RecordBatch]c  cst|jjdtdd} ||jj} | | }|j}| dur$| n| dk} t ||| | d}t |j||| |j }d}|t urUt ||||d}t|||d}t||}t||dd }tjj|||se|ndd d |jDd }d }|}|D]P}|t|}|t|}|}|rt|||t|}||}|jd krqx|durtj|g}||}|jd krqx|d }t|||| |ddVqxWddS1swYdS)NTrXrr)r@projected_field_valuesr@F)select_full_typescSrrr)rLcolrrrrNUrz+_task_to_record_batches..)rirrocolumnsr)rr8allow_timestamp_tz_mismatch) rcr rrrorgrrtphysical_schemarr;r/rr%r#rMrcr^rurvrI to_batchesrrtakenum_rowsrTable from_batchesrocombine_chunks_to_requested_schema)rdr<r=r(r)r>rr@rr*rr arrow_formatfinrirK file_schemar8pyarrow_filtertranslated_row_filterbound_file_filterfile_project_schemafragment_scanner next_indexbatchesbatch current_index current_batchindicesrrrrr_task_to_record_batches#sl            "ratasksIterable[FileScanTask]dict[str, list[ChunkedArray]]c si}ttjdd|D}t|dkrGt}|ddfdd|D}|D]}|D]\}}||vr@|| |q0|g||<q0q*|S)NcSrr) delete_files)rLr<rrrrNrz*_read_all_delete_files..rcSst|Sr)r)rrrrrQsz(_read_all_delete_files..csg|]}|fqSrr)rLrurdrrrNs) r6rr from_iterablerr get_or_createrr`r) rdrbdeletes_per_fileunique_deletesexecutordeletes_per_filesrwr rrrfr_read_all_delete_files}s  rmc@seZdZUded<ded<ded<ded<d ed <d ed <d ed<  d*d+ddZed,ddZd-d!d"Zd.d$d%Zd/d(d)Z dS)0 ArrowScanrj_table_metadatarM_ior\_projected_schemar_bound_row_filterr_case_sensitiver_limitrArTNtable_metadatardr( row_filterr@limitrrscCsD||_||_||_t|||d|_||_||_t t |_ dS)NrE) rorprqr#rrrrsrtrget_boolrgr)rrurdr(rvr@rwrrrrs zArrowScan.__init__r-cs"fddjjDtjS)z>Set of field IDs that should be projected from the data files.cs&h|]}tj|ttfs|qSr)rtrq find_typerr})rLidrrr s z1ArrowScan._projected_field_ids..)rqr/unionr$rrrrrr_projected_field_idss  zArrowScan._projected_field_idsrbrcpa.TablecCsft|jdd}||}zt|}Wn ty|YSwtjddt |g|Ddd}|S)aScan the Iceberg table and return a pa.Table. Returns a pa.Table with data from the Iceberg table by resolving the right columns that match the current table schema. Only data that matches the provided row_filter expression is returned. Args: tasks: FileScanTasks representing the data files and delete files to read from. Returns: A PyArrow table. Total number of rows will be capped if specified. Raises: ResolveError: When a required field cannot be found in the file ValueError: When a field type in the file cannot be projected to the schema type Frcss|] }tj|gVqdSr)rrOrPrLr]rrr sz%ArrowScan.to_table.. permissive)promote_options) rrqto_record_batchesnext StopIteration empty_tabler concat_tablesrr)rrb arrow_schemar\ first_batchrrrrrws    zArrowScan.to_tablerBc #stj|d}t}d fdd }d}|||D]1}|D]'}t|}jd urA||jkrA|dj|Vd }n|V||7}q!|rNd Sqd S) aScan the Iceberg table and return an Iterator[pa.RecordBatch]. Returns an Iterator of pa.RecordBatch with data from the Iceberg table by resolving the right columns that match the current table schema. Only data that matches the provided row_filter expression is returned. Args: tasks: FileScanTasks representing the data files and delete files to read from. Returns: An Iterator of PyArrow RecordBatches. Total number of rows will be capped if specified. Raises: ResolveError: When a required field cannot be found in the file ValueError: When a field type in the file cannot be projected to the schema type rr<rrlist[pa.RecordBatch]cst|gSr)r+_record_batches_from_scan_tasks_and_deletes)r<rirrrbatches_for_tasksz5ArrowScan.to_record_batches..batches_for_taskFNT)r<rrr)rmrprrhrrrtslice) rrbtotal_row_countrkr limit_reachedr\r]current_batch_sizerrrrs&  zArrowScan.to_record_batchesrirdccsd}|D]e}|jdur||jkrdSt|j||j|j|j|j||j j |j |j |j |j j|jj|j }|D])}|jdur`||jkrNn|t||jkr`|d|j|}|V|t|7}q@qdS)Nr)rtrarprrrqrorr}rr rgrsrspecsspec_idrrrr)rrbrirr<r\r]rrrr s:   z5ArrowScan._record_batches_from_scan_tasks_and_deletes)TN)rurjrdrMr(r\rvrr@rrwrrrs)rr-)rbrcrr~)rbrcrrB)rbrcrirdrrB) rrrrrpropertyr}rwrrrrrrrns"     ".rnrequested_schemarUr]pa.RecordBatchr8rJc Cs,t||t|||||dt|}tj|S)N)r8rJ)reArrowProjectionVisitor ArrowAccessorr RecordBatchfrom_struct_array)rrUr]rrr8rJ struct_arrayrrrrR,s  rRc@seZdZUded<ded<ded<ded<ded<d d ed fdAddZdBddZdCddZdDd"d#ZdEd)d*Z dFd-d.Z dGd3d4Z dHd:d;Z dId>d?Z d@S)Jrr\ _file_schemarrrr._projected_missing_fields_allow_timestamp_tz_mismatchFrUrrr8rJrrscCs"||_||_||_||_||_dSr)rrrrr)rrUrrr8rJrrrrLs  zArrowProjectionVisitor.__init__rrvaluesrc Cs|j|j}|jjrt|j|jd}|jkr|jtkro|jj dup,|j o,|jj t v}t j |rd|j sdt j |jrd|rd|jdkrT|jjdkrT|jrT|j|ddS|jdkrd|jjdvrd||Std|jd||jtkrt j |r|j d krt j |jr|jj t vs|jj dur|jdkr|jjdkr|jr|j|ddS|jdkr|jjdvr||Std|jd|t|jttfrt j |jr|jj}|j}||kr||S|j|jkrtt|j|j|jd}||S|S) NrrrF)safe>rrrz#Unsupported schema projection from z to r)r find_fieldrr is_primitiverrrrrrrrrrrrrrrrtr|r~rrrb) rrr file_field target_typesource_tz_compatible source_width target_width target_schemarrr_cast_if_needed\sX             z&ArrowProjectionVisitor._cast_if_needed arrow_typerrcCs>i}|jr |j|t<|jrt|j|t<tj|j||j |dSr) rrrrrrrrrr)rrrrrrr_construct_fields z'ArrowProjectionVisitor._construct_fieldrschema_partnerpa.Array | NonercC|Srr)rrrrrrrrrHzArrowProjectionVisitor.schemarrrrbuiltins.list[pa.Array | None]c CsL|durdSg}g}t|j|ddD]{\}}|dur0|||}||||||jq|js8|jdurt|j |j d} |j |j } rZ|ttj| | dt|n#|jdurl|tjt|| dn|ttj|j| dt||||| qtd|tjj|t|t|tjr|dSddS)NTstrictrrz7Field is required, and could not be found in the file: )arraysrmask)ziprrrrrrinitial_defaultrrrrrrrrepeatrrnullsr StructArray from_arraysrrtr) rrrr field_arraysrr field_arrayrrprojected_valuerrrrs2  " "zArrowProjectionVisitor.structrrcCrrr)rrrrrrrrrHzArrowProjectionVisitor.fieldrr} list_array value_arraycCst|tjtjtjfr?|dur?t|tjrtjntj}t|tjr)tj|j |}| |j |}|| |j |j }||SdSr)rtr ListArrayLargeListArrayFixedSizeListArrayrr'rroffsetsrrrrr)rrrrlist_initializer arrow_fieldrrrrs  zArrowProjectionVisitor.listrr map_arrayrrcCst|tjrC|durC|durC||j|}||j|}t||j|j||j|j}t|tj r>tj |j |||S| |SdSr) rtrMapArrayrrrrrrrrrr)rrrrrrrrrrs  zArrowProjectionVisitor.maprrcCrrr)rrrrrrrrHz ArrowProjectionVisitor.primitiveN) rUr\rrrrr8r.rJrrrs)rrrrrr)rrrrrr)rr\rrrrrr)rrrrrrrr)rrrrrrrr)rr}rrrrrr) rrrrrrrrrr)rrrrrr)rrrrrprrrrrrrrrrrrrrEs&   4     rc@sReZdZUded<dddZddd ZdddZdddZdddZdddZ dS)rr\rUcCrr)rU)rrUrrrrrzArrowAccessor.__init__partnerrrcCs|Srr)rrrrrrrHzArrowAccessor.schema_partnerpartner_structrrrrcCs|durGz |j|j}Wn tyYdSwt|tjr#||St|tjr0| | St|tj r;| |Std|dt |dS)Nz Cannot find z! in expected partner_struct type ) rUrrrrtrrrrOrprQrr)rrrrrrrr field_partners      zArrowAccessor.field_partner partner_listcCs t|tjtjtjfr|jSdSr)rtrrrrr)rrrrrlist_element_partner z"ArrowAccessor.list_element_partner partner_mapcCt|tjr |jSdSr)rtrrkeysrrrrrmap_key_partnerzArrowAccessor.map_key_partnercCrr)rtrrr`rrrrmap_value_partner rzArrowAccessor.map_value_partnerN)rUr\)rrrr)rrrrrrrr)rrrr)rrrr) rrrrrrrrrrrrrrrs      rrcCs t|tSr)r&#_PRIMITIVE_TO_PHYSICAL_TYPE_VISITOR)rrrr_primitive_to_physicalrrc@seZdZdbddZdcd d ZddddZdeddZdfddZdgd d!Zdhd$d%Z did(d)Z djd,d-Z dkd0d1Z dld4d5Z dmd8d9Zdnd.rrrrrrrrz!PyArrowStatisticsCollector.structrrrcC|j|_|Sr)rrrrrrrz PyArrowStatisticsCollector.fieldrr}rcCr9r)rrrrrrrr:zPyArrowStatisticsCollector.listrrrrcCs$|j|_|}|j|_|}||Sr)rrrrrrrrZr[rrrrs zPyArrowStatisticsCollector.maprrcCsddlm}|j|j}|durgSt|j}|j|j d|}|r*t|}t |t s?t |t s?|j tjkr?ttj}|ddk}|rV|j tjtjfvrVttj}t|j|||dgS)Nrr,r)rrrr&)r-rhrrrr$r+r*rMETRICS_MODE_COLUMN_CONF_PREFIXrtrrtrrrrrfindrr%)rrrhr& metrics_modecol_moderrrrr s$     z$PyArrowStatisticsCollector.primitiveN)rr\rr))rr\rr0rr1)rrrr3rr1)rrrr0rr1)rr}rr0rr1)rrrr0rr0rr1)rrrr1 rrrrrrrrrrrrrrrrr(s       r(table_propertiesr)dict[int, StatisticsCollector]cCs,t|t||}i}|D]}|||j<q |S)aQ Compute the statistics plan for all columns. The resulting list is assumed to have the same length and same order as the columns in the pyarrow table. This allows the list to map from the column index to the Iceberg column ID. For each element, the desired metrics collection that was provided by the user in the configuration is computed and then adjusted according to the data type of the column. For nested columns the minimum and maximum values are not computed. And truncation is only applied to text of binary strings. Args: table_properties (from pyiceberg.table.metadata.TableMetadata): The Iceberg table metadata properties. They are required to compute the mapping of column position to iceberg schema type id. It's also used to set the mode for column metrics collection )rar(r)rrA stats_colsr stats_colrrrcompute_statistics_plan s  rEc@eZdZUded<ded<dS)ID2ParquetPathrrr parquet_pathNr'rrrrrG7   rGc@sheZdZUdZded<ded<d,dd Zd-ddZd.ddZd/ddZd0dd Z d1d%d&Z d2d)d*Z d+S)3ID2ParquetPathVisitorrrrrrrrscCs g|_dSr)rrrrrrA rzID2ParquetPathVisitor.__init__rr\r+Callable[[], builtins.list[ID2ParquetPath]]builtins.list[ID2ParquetPath]cCr2rrrrrrrD rizID2ParquetPathVisitor.schemarrr:builtins.list[Callable[[], builtins.list[ID2ParquetPath]]]cCr4)NcSr5rrr6rrrrNJ rz0ID2ParquetPathVisitor.struct..r7rrrrrG r8zID2ParquetPathVisitor.structrrrcCs*|j|_|j|j|}|j|Sr)rrrrrr)rrrrrrrrL s  zID2ParquetPathVisitor.fieldrr}rcCs(|j|_|jd|}|j|S)Nz list.element)rrrrr)rrrrrrrrU s   zID2ParquetPathVisitor.listrrrrcCsP|j|_|jd|}|j|j|_|jd|}|j||S)Nz key_value.keyzkey_value.value)rrrrrrr;rrrr^ s    zID2ParquetPathVisitor.maprrcCst|jd|jdgS)Nr)rrH)rGrrrrrrrrn rzID2ParquetPathVisitor.primitiveNrL)rr\rrKrrL)rrrrMrrL)rrrrKrrL)rr}rrKrrL)rrrrKrrKrrL)rrrrLr@rrrrrJ= s      rJdict[str, int]cCs&i}t|tD]}|j||j<q|S)ax Compute the mapping of parquet column path to Iceberg ID. For each column, the parquet file metadata has a path_in_schema attribute that follows a specific naming scheme for nested columns. This function computes a mapping of the full paths to the corresponding Iceberg IDs. Args: schema (pyiceberg.schema.Schema): The current table schema. )rarJrrH)rrpairrrrparquet_path_to_id_mappingr s rPc@sdeZdZUded<ded<ded<ded<ded<ded <d ed <dddZdddZdddZdS) DataFileStatisticsr record_countzdict[int, int] column_sizes value_countsnull_value_countsnan_value_countszdict[int, StatsAggregator]column_aggregatesz list[int] split_offsetsr9rUrr\rrcCs|j|jvrdS||j}|j}|js td|jd|j||j}|t||j|jj |d}|t||j|jj |d}||krTtd|jd|d||S)NzUCannot infer partition value from parquet metadata for a non-linear Partition Field: z with transform )r9rrztCannot infer partition value from parquet metadata as there are more than one partition values for Partition Field: z. lower_value=z, upper_value=) source_idrWrr3preserves_orderrrrrYrr)rr9r source_fieldiceberg_transformtransform_func lower_value upper_valuerrr_partition_value sL     z#DataFileStatistics._partition_valuer*rXrrcstfdd|jDS)Ncsg|]}|qSr)r`rrrrrrN rOz0DataFileStatistics.partition..)rrr)rr*rrrarr4 r#zDataFileStatistics.partitionrxc Csni}i}|jD]\}}|}|dur|||<|}|dur%|||<q |j|j|j|j|j|||j dS)N)rRrSrTrUrV lower_bounds upper_boundsrX) rWr`r rrRrSrTrUrVrX)rrbrcrZagg_min_maxrrrto_serialized_dict s&z%DataFileStatistics.to_serialized_dictN)r9rUrr\rr)r*rXrr\rrrr~)rrrrr`r4rgrrrrrQ s   %rQparquet_metadatapq.FileMetaData stats_columnsparquet_column_mappingc Csi}i}g}i}i}i}t} t|jD]} || } | dj} | dj} | djr8| | kr8|| n|| t|j D]}| |}||j }||}| |d|||j 7<|j ttjkrjqB||d|j||<|jr%z|j}|jr||d|j||<|j ttjkrWqB||vrzt|j|j|j j||<Wnty}z t|d|jd|d}~wwt|jtr|jdkr|jj }|j!dur||"t#|j!|nd|j$dur||%t#|j$|ndn||"|j&||%|j'WqBt(j)j*y$}z| +|t,-|WYd}~qBd}~ww| +|t,-d|qBq|.| D]}|/|d|/|dq8t0|j1||||||dS)aF Compute and return DataFileStatistics that includes the following. - record_count - column_sizes - value_counts - null_value_counts - nan_value_counts - column_aggregates - split_offsets Args: parquet_metadata (pyarrow.parquet.FileMetaData): A pyarrow metadata object. stats_columns (Dict[int, StatisticsCollector]): The statistics gathering plan. It is required to set the mode for column metrics collection parquet_column_mapping (Dict[str, int]): The mapping of the parquet file name to the field ID rz for column ''Nrz:PyArrow statistics missing for column %d when writing file)rRrSrTrUrVrWrX)2r6rnum_row_groups row_grouprpdata_page_offsetdictionary_page_offsethas_dictionary_pager num_columnspath_in_schema setdefaulttotal_compressed_sizerrrrr num_values is_stats_set statisticshas_null_count null_countrrr physical_typerrr&rtrwrmin_rawrrmax_rawr rrrRlibArrowNotImplementedErrorr9rrsortrrQrN)rhrjrkrSrTrXrUrVcol_aggsinvalidate_colrrn data_offsetdictionary_offsetrlrprrDrxrrrrr*data_file_statistics_from_parquet_metadata s             7 rrurjIterator[WriteTask]Iterator[DataFile]csrddlmm}tjtj|j|jdtj jdd fd d }t }| ||}t |S) Nrrfr property_namedefault)table_locationrAr<rrrRc sN}t|}|kr|n|tpdfddjD}tj|}j dj d} |}|j dd,}t j|f|jdd}|j|d Wdn1sawYWdn1spwYt|jjtjtd } tjd tj|tjj rj jntt|djddd | } | S) NFc s g|] }tj|ddqS)T)rrUr]rr)rRrrrrUr<rrrNZ sz5write_file..write_parquet..parquet)data_file_name partition_keyT)r)rstore_decimal_as_integer)row_group_sizerhrjrk contentrgrr4file_size_in_bytes sort_order_idr equality_ids key_metadatar)!rrdrrxrecord_batchesrrOrPnew_data_locationgenerate_data_file_filenamerrqrpq ParquetWriterwriterwriterrrErrPrR from_argsrSDATArTr]r4rrrdefault_spec_idrg) r<r)sanitized_schemar\ arrow_tablergfofosrrxrergrdlocation_providerparquet_writer_kwargsrrurr write_parquetP s\   z!write_file..write_parquet)r<rrrR)r-rgrh_get_parquet_writer_kwargsrrPARQUET_ROW_GROUP_LIMITPARQUET_ROW_GROUP_LIMIT_DEFAULTrirrrhriter)rdrurbrhrrk data_filesrrr write_fileE s 7 rtblr~target_file_sizeIterator[list[pa.RecordBatch]]cCsTddlm}|j|j}tdt||}|j|d}|||t|dddd}|S) Nr)PackingIteratorr) max_chunksizecSs|jSr)nbytes)xrrrrQ sz&bin_pack_arrow_table..F)r` target_weightlookback weight_funclargest_bin_first)pyiceberg.utils.bin_packingrrrNrrrLr)rrravg_row_size_bytestarget_rows_per_filer\bin_packed_record_batchesrrrbin_pack_arrow_table s   rprovided_schemarsc Cs|j}z t||||d}Wn/ty<}z#t|||d}t|jt|j}tddt|d|d}~wwt ||dS)z Check if the `requested_schema` is compatible with `provided_schema`. Two schemas are considered compatible when they are equal in terms of the Iceberg Schema type. Raises: ValueError: If the schemas are not compatible. rrrrz%PyArrow table contains more columns: z, z4. Update the schema first (hint, use union_by_name).N) rrrrr6 _name_to_idrrrSr_)rrrrrradditional_namesrrr _check_pyarrow_schema_compatible s*  r file_paths Iterator[str]ccs$|D] }t|||d}|VqdS)N)rdrurg)parquet_file_to_data_file)rdrurrgrerrrparquet_files_to_data_files s rrgc Cs||}| }t|}Wdn1swY|j}|}t|||jdt|t ||j t |d}t j dtj|tj|||t|d|jddd |} | S)N)rrrr)rorr read_metadatarto_arrow_schemarrrrErrPrRrrSrrTr]r4specrrrg) rdrurgr input_streamrhrrrxrerrrr s4       r uncompressedrrqc Csddlm}|j|j|jdfD]}t||}r&tjd|dddq| |j |j }t ||j |jd}|tkr>t}||t ||j|jdt ||j|jdt ||j|jdd S) Nrr,z.*zParquet writer option(s) z not implementedrrr) compressioncompression_leveldata_page_sizedictionary_pagesize_limitwrite_batch_size)r-rhPARQUET_ROW_GROUP_SIZE_BYTESPARQUET_BLOOM_FILTER_MAX_BYTES*PARQUET_BLOOM_FILTER_COLUMN_ENABLED_PREFIXfnmatchrorrrPARQUET_COMPRESSIONPARQUET_COMPRESSION_DEFAULTrPARQUET_COMPRESSION_LEVEL!PARQUET_COMPRESSION_LEVEL_DEFAULTICEBERG_UNCOMPRESSED_CODECPYARROW_UNCOMPRESSED_CODECPARQUET_PAGE_SIZE_BYTESPARQUET_PAGE_SIZE_BYTES_DEFAULTPARQUET_DICT_SIZE_BYTESPARQUET_DICT_SIZE_BYTES_DEFAULTPARQUET_PAGE_ROW_LIMITPARQUET_PAGE_ROW_LIMIT_DEFAULT)rArh key_patternunsupported_keyscompression_codecrrrrr sD  rdf write_uuiduuid.UUID | Nonecounteritertools.count[int] | NoneIterable[DataFile]c #sddlm}m}mptdptt|j |j |j d| j }t|p.d}t|j |||jd|rXt||fddt|DdEd Hd St|| |d } t||fd d| DdEd Hd S) zConvert a PyArrow table into a DataFile. Returns: An iterable that supplies datafiles that represent the table. r)rgrhrrFrc3s$|] }t|dVqdS))rtask_idrrN)r)rLr\)rr task_schemarrrr4 s  z+_dataframe_to_data_files..)rdrurbN)rrrc 3s:|]}t|jD]}t||jdVq qdS))rrrrrN)rarrow_table_partitionrr)rLr4r\rrrrrrrr> s   )r-rgrhrrcountruuid4rrWRITE_TARGET_FILE_SIZE_BYTES$WRITE_TARGET_FILE_SIZE_BYTES_DEFAULTrrrrxrrris_unpartitionedrr_determine_partitions) rurrdrrrgrhrr partitionsrrr_dataframe_to_data_files sB     rc@rF)_TablePartitionrWrr~rNr'rrrrrL rIrrrXrIterable[_TablePartition]c #sdd|jD}t|j|ddD].\}}||j}||j}|dur-td|jt||}|||j |j |}q| | | g} | D];tfddt|j|ddD||d} |ttjfd dt|j|ddD} | |} t| | d VqOdS) aBased on the iceberg table partition spec, filter the arrow table into partitions with their keys. Example: Input: An arrow table with partition key of ['n_legs', 'year'] and with data of {'year': [2020, 2022, 2022, 2021, 2022, 2022, 2022, 2019, 2021], 'n_legs': [2, 2, 2, 4, 4, 4, 4, 5, 100], 'animal': ["Flamingo", "Parrot", "Parrot", "Dog", "Horse", "Horse", "Horse","Brittle stars", "Centipede"]}. The algorithm: - We determine the set of unique partition keys - Then we produce a set of partitions by filtering on each of the combinations - We combine the chunks to create a copy to avoid GIL congestion on the original table cSsg|]}d|jqS) _partition_rGrrrrrNb rOz)_determine_partitions..TrNz)Could not find column name for field ID: cs g|] \}}t||dqS))rr)rV)rLrrunique_partitionrrrNp s) field_valuesr*rcs<g|]\}}|durt||knt|qSr)rrr)rLrpartition_field_namerrrrNz s   )rr)rrrrYrr_get_field_from_arrow_table append_columnr3pyarrow_transformrselectgroup_by aggregate to_pylistrWro functoolsreduceoperatorand_ drop_columnsrrQ) rrrpartition_fieldsr4rr[full_field_namerunique_partition_fieldsrfiltered_tablerrrrR s@        rrcCs<||jvr ||S|d}||d}t||ddS)a Get a field from an Arrow table, supporting both literal field names and nested field paths. This function handles two cases: 1. Literal field names that may contain dots (e.g., "some.id") 2. Nested field paths using dot notation (e.g., "bar.baz" for nested access) Args: arrow_table: The Arrow table containing the field field_path: Field name or dot-separated path Returns: The field as a PyArrow Array Raises: KeyError: If the field path cannot be resolved rrrN) column_namesrr struct_field)rrrrrrrr s   r)rrrr)rrrr) rrrrrrrrTrr)rrrr{rrr)rIrrrrr)rrTrrxrrW)rdrMrerRrrf)rrrrrrrr) rrrrrrrrsrr\)rrrrrrsrr\)rrrr)rrrrrr)rrrrrr)rrrrrr)rrrrrr)rrrrrr)rrrrrr)rrrrrr)rrrrrr)rrrr) r rRr(r\r)r\r*r+r,r-rr.)rdrMr<rr=rr(r\r)r\r>r-rr?r@rrrr*r+rrsrrArrB)rdrMrbrcrrd)rr\rUr\r]rrrrrr8r.rJrrr)rrrr)rrrr)rr\rAr)rrB)rr\rrN)rhrirjrBrkrNrrQ)rdrMrurjrbrrr)rr~rrrr) rr\rrrrrrsrrs)rdrMrurjrrrr)rdrMrurjrgrrrR)rArqrrx)NN) rurjrr~rdrMrrrrrr)rrXrr\rr~rr)rr~rrrr(0r __future__rbuiltinsrrrrloggingr rrrrabcrrcollections.abcrrrr dataclassesr enumr r r typingr rrrr urllib.parserrRrpyarrow.computecomputerpyarrow.datasetdatasetr^ pyarrow.libpyarrow.parquetrrr pyarrow._s3fsrrrrrpyiceberg.conversionsrpyiceberg.exceptionsrpyiceberg.expressionsrrrrrrr pyiceberg.expressions.literalsr!pyiceberg.expressions.visitorsr"r#r$r%r&rJ pyiceberg.ior'r(r)r*r+r,r-r.r/r0r1r2r3r4r5r6r7r8r9r:r;r<r=r>r?r@rArBrCrDrErFrGrHrIrJrKrLrMrNrOrPrQpyiceberg.manifestrRrSrTpyiceberg.partitioningrUrVrWrXrYpyiceberg.schemarZr[r\r]r^r_r`rarbrcrdrer-rgrhpyiceberg.table.locationsripyiceberg.table.metadatarjpyiceberg.table.name_mappingrkrlpyiceberg.table.puffinrmpyiceberg.transformsrnropyiceberg.typedefrprqrrrspyiceberg.typesrtrurvrwrxryrzr{r|r}r~rrrrrrrrrrrrrpyiceberg.utils.concurrentrpyiceberg.utils.configrpyiceberg.utils.datetimerpyiceberg.utils.decimalrpyiceberg.utils.propertiesrrrpyiceberg.utils.singletonrpyiceberg.utils.truncaterrrr getLoggerrrrrmICEBERG_SCHEMArrrrrrrDOCrrrr ExceptionrrLocalFileSystemrrrr]rDataTyperr Expressionrr1rMrVrcrrrrrrrrregisterrFixedSizeListType LargeListTypeDictionaryTypeFieldrrrrrrrrr;rarmrnrRArrayrrrrrrrDEFAULT_TRUNCATION_LENGTHr!rrr$r%rr(rErGrJrPrQrrrrrrrrrrrrrrrrrsp            $  -<   h       3 l_y           2"Z!&$& DKL5HwH#-:<