Skip to content

hive

HiveCatalog

Bases: MetastoreCatalog

Source code in pyiceberg/catalog/hive.py
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
class HiveCatalog(MetastoreCatalog):
    _client: _HiveClient

    def __init__(self, name: str, **properties: str):
        super().__init__(name, **properties)
        self._client = self._create_hive_client(properties)

        self._lock_check_min_wait_time = property_as_float(properties, LOCK_CHECK_MIN_WAIT_TIME, DEFAULT_LOCK_CHECK_MIN_WAIT_TIME)
        self._lock_check_max_wait_time = property_as_float(properties, LOCK_CHECK_MAX_WAIT_TIME, DEFAULT_LOCK_CHECK_MAX_WAIT_TIME)
        self._lock_check_retries = property_as_float(
            properties,
            LOCK_CHECK_RETRIES,
            DEFAULT_LOCK_CHECK_RETRIES,
        )

    @staticmethod
    def _create_hive_client(properties: Dict[str, str]) -> _HiveClient:
        last_exception = None
        for uri in properties["uri"].split(","):
            try:
                return _HiveClient(uri, properties.get("ugi"))
            except BaseException as e:
                last_exception = e
        if last_exception is not None:
            raise last_exception
        else:
            raise ValueError(f"Unable to connect to hive using uri: {properties['uri']}")

    def _convert_hive_into_iceberg(self, table: HiveTable) -> Table:
        properties: Dict[str, str] = table.parameters
        if TABLE_TYPE not in properties:
            raise NoSuchPropertyException(
                f"Property table_type missing, could not determine type: {table.dbName}.{table.tableName}"
            )

        table_type = properties[TABLE_TYPE]
        if table_type.lower() != ICEBERG:
            raise NoSuchIcebergTableError(
                f"Property table_type is {table_type}, expected {ICEBERG}: {table.dbName}.{table.tableName}"
            )

        if prop_metadata_location := properties.get(METADATA_LOCATION):
            metadata_location = prop_metadata_location
        else:
            raise NoSuchPropertyException(f"Table property {METADATA_LOCATION} is missing")

        io = self._load_file_io(location=metadata_location)
        file = io.new_input(metadata_location)
        metadata = FromInputFile.table_metadata(file)
        return Table(
            identifier=(table.dbName, table.tableName),
            metadata=metadata,
            metadata_location=metadata_location,
            io=self._load_file_io(metadata.properties, metadata_location),
            catalog=self,
        )

    def _convert_iceberg_into_hive(self, table: Table) -> HiveTable:
        identifier_tuple = self._identifier_to_tuple_without_catalog(table.identifier)
        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
        current_time_millis = int(time.time() * 1000)

        return HiveTable(
            dbName=database_name,
            tableName=table_name,
            owner=table.properties[OWNER] if table.properties and OWNER in table.properties else getpass.getuser(),
            createTime=current_time_millis // 1000,
            lastAccessTime=current_time_millis // 1000,
            sd=_construct_hive_storage_descriptor(
                table.schema(),
                table.location(),
                property_as_bool(self.properties, HIVE2_COMPATIBLE, HIVE2_COMPATIBLE_DEFAULT),
            ),
            tableType=EXTERNAL_TABLE,
            parameters=_construct_parameters(table.metadata_location),
        )

    def _create_hive_table(self, open_client: Client, hive_table: HiveTable) -> None:
        try:
            open_client.create_table(hive_table)
        except AlreadyExistsException as e:
            raise TableAlreadyExistsError(f"Table {hive_table.dbName}.{hive_table.tableName} already exists") from e

    def _get_hive_table(self, open_client: Client, database_name: str, table_name: str) -> HiveTable:
        try:
            return open_client.get_table(dbname=database_name, tbl_name=table_name)
        except NoSuchObjectException as e:
            raise NoSuchTableError(f"Table does not exists: {table_name}") from e

    def create_table(
        self,
        identifier: Union[str, Identifier],
        schema: Union[Schema, "pa.Schema"],
        location: Optional[str] = None,
        partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
        sort_order: SortOrder = UNSORTED_SORT_ORDER,
        properties: Properties = EMPTY_DICT,
    ) -> Table:
        """Create a table.

        Args:
            identifier: Table identifier.
            schema: Table's schema.
            location: Location for the table. Optional Argument.
            partition_spec: PartitionSpec for the table.
            sort_order: SortOrder for the table.
            properties: Table properties that can be a string based dictionary.

        Returns:
            Table: the created table instance.

        Raises:
            AlreadyExistsError: If a table with the name already exists.
            ValueError: If the identifier is invalid.
        """
        properties = {**DEFAULT_PROPERTIES, **properties}
        staged_table = self._create_staged_table(
            identifier=identifier,
            schema=schema,
            location=location,
            partition_spec=partition_spec,
            sort_order=sort_order,
            properties=properties,
        )
        database_name, table_name = self.identifier_to_database_and_table(identifier)

        self._write_metadata(staged_table.metadata, staged_table.io, staged_table.metadata_location)
        tbl = self._convert_iceberg_into_hive(staged_table)

        with self._client as open_client:
            self._create_hive_table(open_client, tbl)
            hive_table = open_client.get_table(dbname=database_name, tbl_name=table_name)

        return self._convert_hive_into_iceberg(hive_table)

    def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
        """Register a new table using existing metadata.

        Args:
            identifier Union[str, Identifier]: Table identifier for the table
            metadata_location str: The location to the metadata

        Returns:
            Table: The newly registered table

        Raises:
            TableAlreadyExistsError: If the table already exists
        """
        raise NotImplementedError

    def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
        raise NotImplementedError

    def _create_lock_request(self, database_name: str, table_name: str) -> LockRequest:
        lock_component: LockComponent = LockComponent(
            level=LockLevel.TABLE, type=LockType.EXCLUSIVE, dbname=database_name, tablename=table_name, isTransactional=True
        )

        lock_request: LockRequest = LockRequest(component=[lock_component], user=getpass.getuser(), hostname=socket.gethostname())

        return lock_request

    def _wait_for_lock(self, database_name: str, table_name: str, lockid: int, open_client: Client) -> LockResponse:
        @retry(
            retry=retry_if_exception_type(WaitingForLockException),
            wait=wait_exponential(multiplier=2, min=self._lock_check_min_wait_time, max=self._lock_check_max_wait_time),
            stop=stop_after_attempt(self._lock_check_retries),
            reraise=True,
        )
        def _do_wait_for_lock() -> LockResponse:
            response: LockResponse = open_client.check_lock(CheckLockRequest(lockid=lockid))
            if response.state == LockState.ACQUIRED:
                return response
            elif response.state == LockState.WAITING:
                msg = f"Wait on lock for {database_name}.{table_name}"
                logger.warning(msg)
                raise WaitingForLockException(msg)
            else:
                raise CommitFailedException(f"Failed to check lock for {database_name}.{table_name}, state: {response.state}")

        return _do_wait_for_lock()

    def commit_table(
        self, table: Table, requirements: Tuple[TableRequirement, ...], updates: Tuple[TableUpdate, ...]
    ) -> CommitTableResponse:
        """Commit updates to a table.

        Args:
            table (Table): The table to be updated.
            requirements: (Tuple[TableRequirement, ...]): Table requirements.
            updates: (Tuple[TableUpdate, ...]): Table updates.

        Returns:
            CommitTableResponse: The updated metadata.

        Raises:
            NoSuchTableError: If a table with the given identifier does not exist.
            CommitFailedException: Requirement not met, or a conflict with a concurrent commit.
        """
        table_identifier = self._identifier_to_tuple_without_catalog(table.identifier)
        database_name, table_name = self.identifier_to_database_and_table(table_identifier, NoSuchTableError)
        # commit to hive
        # https://github.com/apache/hive/blob/master/standalone-metastore/metastore-common/src/main/thrift/hive_metastore.thrift#L1232
        with self._client as open_client:
            lock: LockResponse = open_client.lock(self._create_lock_request(database_name, table_name))

            try:
                if lock.state != LockState.ACQUIRED:
                    if lock.state == LockState.WAITING:
                        self._wait_for_lock(database_name, table_name, lock.lockid, open_client)
                    else:
                        raise CommitFailedException(f"Failed to acquire lock for {table_identifier}, state: {lock.state}")

                hive_table: Optional[HiveTable]
                current_table: Optional[Table]
                try:
                    hive_table = self._get_hive_table(open_client, database_name, table_name)
                    current_table = self._convert_hive_into_iceberg(hive_table)
                except NoSuchTableError:
                    hive_table = None
                    current_table = None

                updated_staged_table = self._update_and_stage_table(current_table, table_identifier, requirements, updates)
                if current_table and updated_staged_table.metadata == current_table.metadata:
                    # no changes, do nothing
                    return CommitTableResponse(metadata=current_table.metadata, metadata_location=current_table.metadata_location)
                self._write_metadata(
                    metadata=updated_staged_table.metadata,
                    io=updated_staged_table.io,
                    metadata_path=updated_staged_table.metadata_location,
                )

                if hive_table and current_table:
                    # Table exists, update it.
                    hive_table.parameters = _construct_parameters(
                        metadata_location=updated_staged_table.metadata_location,
                        previous_metadata_location=current_table.metadata_location,
                    )
                    open_client.alter_table(dbname=database_name, tbl_name=table_name, new_tbl=hive_table)
                else:
                    # Table does not exist, create it.
                    hive_table = self._convert_iceberg_into_hive(
                        StagedTable(
                            identifier=(database_name, table_name),
                            metadata=updated_staged_table.metadata,
                            metadata_location=updated_staged_table.metadata_location,
                            io=updated_staged_table.io,
                            catalog=self,
                        )
                    )
                    self._create_hive_table(open_client, hive_table)
            except WaitingForLockException as e:
                raise CommitFailedException(f"Failed to acquire lock for {table_identifier}, state: {lock.state}") from e
            finally:
                open_client.unlock(UnlockRequest(lockid=lock.lockid))

        return CommitTableResponse(
            metadata=updated_staged_table.metadata, metadata_location=updated_staged_table.metadata_location
        )

    def load_table(self, identifier: Union[str, Identifier]) -> Table:
        """Load the table's metadata and return the table instance.

        You can also use this method to check for table existence using 'try catalog.table() except TableNotFoundError'.
        Note: This method doesn't scan data stored in the table.

        Args:
            identifier: Table identifier.

        Returns:
            Table: the table instance with its metadata.

        Raises:
            NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
        """
        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)

        with self._client as open_client:
            hive_table = self._get_hive_table(open_client, database_name, table_name)

        return self._convert_hive_into_iceberg(hive_table)

    def drop_table(self, identifier: Union[str, Identifier]) -> None:
        """Drop a table.

        Args:
            identifier: Table identifier.

        Raises:
            NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
        """
        identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
        try:
            with self._client as open_client:
                open_client.drop_table(dbname=database_name, name=table_name, deleteData=False)
        except NoSuchObjectException as e:
            # When the namespace doesn't exist, it throws the same error
            raise NoSuchTableError(f"Table does not exists: {table_name}") from e

    def purge_table(self, identifier: Union[str, Identifier]) -> None:
        # This requires to traverse the reachability set, and drop all the data files.
        raise NotImplementedError("Not yet implemented")

    def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: Union[str, Identifier]) -> Table:
        """Rename a fully classified table name.

        Args:
            from_identifier: Existing table identifier.
            to_identifier: New table identifier.

        Returns:
            Table: the updated table instance with its metadata.

        Raises:
            ValueError: When from table identifier is invalid.
            NoSuchTableError: When a table with the name does not exist.
            NoSuchNamespaceError: When the destination namespace doesn't exist.
        """
        from_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
        to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
        try:
            with self._client as open_client:
                tbl = open_client.get_table(dbname=from_database_name, tbl_name=from_table_name)
                tbl.dbName = to_database_name
                tbl.tableName = to_table_name
                open_client.alter_table(dbname=from_database_name, tbl_name=from_table_name, new_tbl=tbl)
        except NoSuchObjectException as e:
            raise NoSuchTableError(f"Table does not exist: {from_table_name}") from e
        except InvalidOperationException as e:
            raise NoSuchNamespaceError(f"Database does not exists: {to_database_name}") from e
        return self.load_table(to_identifier)

    def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
        """Create a namespace in the catalog.

        Args:
            namespace: Namespace identifier.
            properties: A string dictionary of properties for the given namespace.

        Raises:
            ValueError: If the identifier is invalid.
            AlreadyExistsError: If a namespace with the given name already exists.
        """
        database_name = self.identifier_to_database(namespace)
        hive_database = HiveDatabase(name=database_name, parameters=properties)

        try:
            with self._client as open_client:
                open_client.create_database(_annotate_namespace(hive_database, properties))
        except AlreadyExistsException as e:
            raise NamespaceAlreadyExistsError(f"Database {database_name} already exists") from e

    def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
        """Drop a namespace.

        Args:
            namespace: Namespace identifier.

        Raises:
            NoSuchNamespaceError: If a namespace with the given name does not exist, or the identifier is invalid.
            NamespaceNotEmptyError: If the namespace is not empty.
        """
        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
        try:
            with self._client as open_client:
                open_client.drop_database(database_name, deleteData=False, cascade=False)
        except InvalidOperationException as e:
            raise NamespaceNotEmptyError(f"Database {database_name} is not empty") from e
        except MetaException as e:
            raise NoSuchNamespaceError(f"Database does not exists: {database_name}") from e

    def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
        """List Iceberg tables under the given namespace in the catalog.

        When the database doesn't exist, it will just return an empty list.

        Args:
            namespace: Database to list.

        Returns:
            List[Identifier]: list of table identifiers.

        Raises:
            NoSuchNamespaceError: If a namespace with the given name does not exist, or the identifier is invalid.
        """
        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
        with self._client as open_client:
            return [
                (database_name, table.tableName)
                for table in open_client.get_table_objects_by_name(
                    dbname=database_name, tbl_names=open_client.get_all_tables(db_name=database_name)
                )
                if table.parameters[TABLE_TYPE].lower() == ICEBERG
            ]

    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
        """List namespaces from the given namespace. If not given, list top-level namespaces from the catalog.

        Returns:
            List[Identifier]: a List of namespace identifiers.
        """
        # Hierarchical namespace is not supported. Return an empty list
        if namespace:
            return []

        with self._client as open_client:
            return list(map(self.identifier_to_tuple, open_client.get_all_databases()))

    def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
        """Get properties for a namespace.

        Args:
            namespace: Namespace identifier.

        Returns:
            Properties: Properties for the given namespace.

        Raises:
            NoSuchNamespaceError: If a namespace with the given name does not exist, or identifier is invalid.
        """
        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
        try:
            with self._client as open_client:
                database = open_client.get_database(name=database_name)
                properties = database.parameters
                properties[LOCATION] = database.locationUri
                if comment := database.description:
                    properties[COMMENT] = comment
                return properties
        except NoSuchObjectException as e:
            raise NoSuchNamespaceError(f"Database does not exists: {database_name}") from e

    def update_namespace_properties(
        self, namespace: Union[str, Identifier], removals: Optional[Set[str]] = None, updates: Properties = EMPTY_DICT
    ) -> PropertiesUpdateSummary:
        """Remove provided property keys and update properties for a namespace.

        Args:
            namespace: Namespace identifier.
            removals: Set of property keys that need to be removed. Optional Argument.
            updates: Properties to be updated for the given namespace.

        Raises:
            NoSuchNamespaceError: If a namespace with the given name does not exist
            ValueError: If removals and updates have overlapping keys.
        """
        self._check_for_overlap(updates=updates, removals=removals)
        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
        with self._client as open_client:
            try:
                database = open_client.get_database(database_name)
                parameters = database.parameters
            except NoSuchObjectException as e:
                raise NoSuchNamespaceError(f"Database does not exists: {database_name}") from e

            removed: Set[str] = set()
            updated: Set[str] = set()

            if removals:
                for key in removals:
                    if key in parameters:
                        parameters[key] = None
                        removed.add(key)
            if updates:
                for key, value in updates.items():
                    parameters[key] = value
                    updated.add(key)

            open_client.alter_database(database_name, _annotate_namespace(database, parameters))

        expected_to_change = (removals or set()).difference(removed)

        return PropertiesUpdateSummary(removed=list(removed or []), updated=list(updated or []), missing=list(expected_to_change))

    def drop_view(self, identifier: Union[str, Identifier]) -> None:
        raise NotImplementedError

commit_table(table, requirements, updates)

Commit updates to a table.

Parameters:

Name Type Description Default
table Table

The table to be updated.

required
requirements Tuple[TableRequirement, ...]

(Tuple[TableRequirement, ...]): Table requirements.

required
updates Tuple[TableUpdate, ...]

(Tuple[TableUpdate, ...]): Table updates.

required

Returns:

Name Type Description
CommitTableResponse CommitTableResponse

The updated metadata.

Raises:

Type Description
NoSuchTableError

If a table with the given identifier does not exist.

CommitFailedException

Requirement not met, or a conflict with a concurrent commit.

Source code in pyiceberg/catalog/hive.py
def commit_table(
    self, table: Table, requirements: Tuple[TableRequirement, ...], updates: Tuple[TableUpdate, ...]
) -> CommitTableResponse:
    """Commit updates to a table.

    Args:
        table (Table): The table to be updated.
        requirements: (Tuple[TableRequirement, ...]): Table requirements.
        updates: (Tuple[TableUpdate, ...]): Table updates.

    Returns:
        CommitTableResponse: The updated metadata.

    Raises:
        NoSuchTableError: If a table with the given identifier does not exist.
        CommitFailedException: Requirement not met, or a conflict with a concurrent commit.
    """
    table_identifier = self._identifier_to_tuple_without_catalog(table.identifier)
    database_name, table_name = self.identifier_to_database_and_table(table_identifier, NoSuchTableError)
    # commit to hive
    # https://github.com/apache/hive/blob/master/standalone-metastore/metastore-common/src/main/thrift/hive_metastore.thrift#L1232
    with self._client as open_client:
        lock: LockResponse = open_client.lock(self._create_lock_request(database_name, table_name))

        try:
            if lock.state != LockState.ACQUIRED:
                if lock.state == LockState.WAITING:
                    self._wait_for_lock(database_name, table_name, lock.lockid, open_client)
                else:
                    raise CommitFailedException(f"Failed to acquire lock for {table_identifier}, state: {lock.state}")

            hive_table: Optional[HiveTable]
            current_table: Optional[Table]
            try:
                hive_table = self._get_hive_table(open_client, database_name, table_name)
                current_table = self._convert_hive_into_iceberg(hive_table)
            except NoSuchTableError:
                hive_table = None
                current_table = None

            updated_staged_table = self._update_and_stage_table(current_table, table_identifier, requirements, updates)
            if current_table and updated_staged_table.metadata == current_table.metadata:
                # no changes, do nothing
                return CommitTableResponse(metadata=current_table.metadata, metadata_location=current_table.metadata_location)
            self._write_metadata(
                metadata=updated_staged_table.metadata,
                io=updated_staged_table.io,
                metadata_path=updated_staged_table.metadata_location,
            )

            if hive_table and current_table:
                # Table exists, update it.
                hive_table.parameters = _construct_parameters(
                    metadata_location=updated_staged_table.metadata_location,
                    previous_metadata_location=current_table.metadata_location,
                )
                open_client.alter_table(dbname=database_name, tbl_name=table_name, new_tbl=hive_table)
            else:
                # Table does not exist, create it.
                hive_table = self._convert_iceberg_into_hive(
                    StagedTable(
                        identifier=(database_name, table_name),
                        metadata=updated_staged_table.metadata,
                        metadata_location=updated_staged_table.metadata_location,
                        io=updated_staged_table.io,
                        catalog=self,
                    )
                )
                self._create_hive_table(open_client, hive_table)
        except WaitingForLockException as e:
            raise CommitFailedException(f"Failed to acquire lock for {table_identifier}, state: {lock.state}") from e
        finally:
            open_client.unlock(UnlockRequest(lockid=lock.lockid))

    return CommitTableResponse(
        metadata=updated_staged_table.metadata, metadata_location=updated_staged_table.metadata_location
    )

create_namespace(namespace, properties=EMPTY_DICT)

Create a namespace in the catalog.

Parameters:

Name Type Description Default
namespace Union[str, Identifier]

Namespace identifier.

required
properties Properties

A string dictionary of properties for the given namespace.

EMPTY_DICT

Raises:

Type Description
ValueError

If the identifier is invalid.

AlreadyExistsError

If a namespace with the given name already exists.

Source code in pyiceberg/catalog/hive.py
def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
    """Create a namespace in the catalog.

    Args:
        namespace: Namespace identifier.
        properties: A string dictionary of properties for the given namespace.

    Raises:
        ValueError: If the identifier is invalid.
        AlreadyExistsError: If a namespace with the given name already exists.
    """
    database_name = self.identifier_to_database(namespace)
    hive_database = HiveDatabase(name=database_name, parameters=properties)

    try:
        with self._client as open_client:
            open_client.create_database(_annotate_namespace(hive_database, properties))
    except AlreadyExistsException as e:
        raise NamespaceAlreadyExistsError(f"Database {database_name} already exists") from e

create_table(identifier, schema, location=None, partition_spec=UNPARTITIONED_PARTITION_SPEC, sort_order=UNSORTED_SORT_ORDER, properties=EMPTY_DICT)

Create a table.

Parameters:

Name Type Description Default
identifier Union[str, Identifier]

Table identifier.

required
schema Union[Schema, Schema]

Table's schema.

required
location Optional[str]

Location for the table. Optional Argument.

None
partition_spec PartitionSpec

PartitionSpec for the table.

UNPARTITIONED_PARTITION_SPEC
sort_order SortOrder

SortOrder for the table.

UNSORTED_SORT_ORDER
properties Properties

Table properties that can be a string based dictionary.

EMPTY_DICT

Returns:

Name Type Description
Table Table

the created table instance.

Raises:

Type Description
AlreadyExistsError

If a table with the name already exists.

ValueError

If the identifier is invalid.

Source code in pyiceberg/catalog/hive.py
def create_table(
    self,
    identifier: Union[str, Identifier],
    schema: Union[Schema, "pa.Schema"],
    location: Optional[str] = None,
    partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
    sort_order: SortOrder = UNSORTED_SORT_ORDER,
    properties: Properties = EMPTY_DICT,
) -> Table:
    """Create a table.

    Args:
        identifier: Table identifier.
        schema: Table's schema.
        location: Location for the table. Optional Argument.
        partition_spec: PartitionSpec for the table.
        sort_order: SortOrder for the table.
        properties: Table properties that can be a string based dictionary.

    Returns:
        Table: the created table instance.

    Raises:
        AlreadyExistsError: If a table with the name already exists.
        ValueError: If the identifier is invalid.
    """
    properties = {**DEFAULT_PROPERTIES, **properties}
    staged_table = self._create_staged_table(
        identifier=identifier,
        schema=schema,
        location=location,
        partition_spec=partition_spec,
        sort_order=sort_order,
        properties=properties,
    )
    database_name, table_name = self.identifier_to_database_and_table(identifier)

    self._write_metadata(staged_table.metadata, staged_table.io, staged_table.metadata_location)
    tbl = self._convert_iceberg_into_hive(staged_table)

    with self._client as open_client:
        self._create_hive_table(open_client, tbl)
        hive_table = open_client.get_table(dbname=database_name, tbl_name=table_name)

    return self._convert_hive_into_iceberg(hive_table)

drop_namespace(namespace)

Drop a namespace.

Parameters:

Name Type Description Default
namespace Union[str, Identifier]

Namespace identifier.

required

Raises:

Type Description
NoSuchNamespaceError

If a namespace with the given name does not exist, or the identifier is invalid.

NamespaceNotEmptyError

If the namespace is not empty.

Source code in pyiceberg/catalog/hive.py
def drop_namespace(self, namespace: Union[str, Identifier]) -> None:
    """Drop a namespace.

    Args:
        namespace: Namespace identifier.

    Raises:
        NoSuchNamespaceError: If a namespace with the given name does not exist, or the identifier is invalid.
        NamespaceNotEmptyError: If the namespace is not empty.
    """
    database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
    try:
        with self._client as open_client:
            open_client.drop_database(database_name, deleteData=False, cascade=False)
    except InvalidOperationException as e:
        raise NamespaceNotEmptyError(f"Database {database_name} is not empty") from e
    except MetaException as e:
        raise NoSuchNamespaceError(f"Database does not exists: {database_name}") from e

drop_table(identifier)

Drop a table.

Parameters:

Name Type Description Default
identifier Union[str, Identifier]

Table identifier.

required

Raises:

Type Description
NoSuchTableError

If a table with the name does not exist, or the identifier is invalid.

Source code in pyiceberg/catalog/hive.py
def drop_table(self, identifier: Union[str, Identifier]) -> None:
    """Drop a table.

    Args:
        identifier: Table identifier.

    Raises:
        NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
    """
    identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
    database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
    try:
        with self._client as open_client:
            open_client.drop_table(dbname=database_name, name=table_name, deleteData=False)
    except NoSuchObjectException as e:
        # When the namespace doesn't exist, it throws the same error
        raise NoSuchTableError(f"Table does not exists: {table_name}") from e

list_namespaces(namespace=())

List namespaces from the given namespace. If not given, list top-level namespaces from the catalog.

Returns:

Type Description
List[Identifier]

List[Identifier]: a List of namespace identifiers.

Source code in pyiceberg/catalog/hive.py
def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
    """List namespaces from the given namespace. If not given, list top-level namespaces from the catalog.

    Returns:
        List[Identifier]: a List of namespace identifiers.
    """
    # Hierarchical namespace is not supported. Return an empty list
    if namespace:
        return []

    with self._client as open_client:
        return list(map(self.identifier_to_tuple, open_client.get_all_databases()))

list_tables(namespace)

List Iceberg tables under the given namespace in the catalog.

When the database doesn't exist, it will just return an empty list.

Parameters:

Name Type Description Default
namespace Union[str, Identifier]

Database to list.

required

Returns:

Type Description
List[Identifier]

List[Identifier]: list of table identifiers.

Raises:

Type Description
NoSuchNamespaceError

If a namespace with the given name does not exist, or the identifier is invalid.

Source code in pyiceberg/catalog/hive.py
def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
    """List Iceberg tables under the given namespace in the catalog.

    When the database doesn't exist, it will just return an empty list.

    Args:
        namespace: Database to list.

    Returns:
        List[Identifier]: list of table identifiers.

    Raises:
        NoSuchNamespaceError: If a namespace with the given name does not exist, or the identifier is invalid.
    """
    database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
    with self._client as open_client:
        return [
            (database_name, table.tableName)
            for table in open_client.get_table_objects_by_name(
                dbname=database_name, tbl_names=open_client.get_all_tables(db_name=database_name)
            )
            if table.parameters[TABLE_TYPE].lower() == ICEBERG
        ]

load_namespace_properties(namespace)

Get properties for a namespace.

Parameters:

Name Type Description Default
namespace Union[str, Identifier]

Namespace identifier.

required

Returns:

Name Type Description
Properties Properties

Properties for the given namespace.

Raises:

Type Description
NoSuchNamespaceError

If a namespace with the given name does not exist, or identifier is invalid.

Source code in pyiceberg/catalog/hive.py
def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
    """Get properties for a namespace.

    Args:
        namespace: Namespace identifier.

    Returns:
        Properties: Properties for the given namespace.

    Raises:
        NoSuchNamespaceError: If a namespace with the given name does not exist, or identifier is invalid.
    """
    database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
    try:
        with self._client as open_client:
            database = open_client.get_database(name=database_name)
            properties = database.parameters
            properties[LOCATION] = database.locationUri
            if comment := database.description:
                properties[COMMENT] = comment
            return properties
    except NoSuchObjectException as e:
        raise NoSuchNamespaceError(f"Database does not exists: {database_name}") from e

load_table(identifier)

Load the table's metadata and return the table instance.

You can also use this method to check for table existence using 'try catalog.table() except TableNotFoundError'. Note: This method doesn't scan data stored in the table.

Parameters:

Name Type Description Default
identifier Union[str, Identifier]

Table identifier.

required

Returns:

Name Type Description
Table Table

the table instance with its metadata.

Raises:

Type Description
NoSuchTableError

If a table with the name does not exist, or the identifier is invalid.

Source code in pyiceberg/catalog/hive.py
def load_table(self, identifier: Union[str, Identifier]) -> Table:
    """Load the table's metadata and return the table instance.

    You can also use this method to check for table existence using 'try catalog.table() except TableNotFoundError'.
    Note: This method doesn't scan data stored in the table.

    Args:
        identifier: Table identifier.

    Returns:
        Table: the table instance with its metadata.

    Raises:
        NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
    """
    identifier_tuple = self._identifier_to_tuple_without_catalog(identifier)
    database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)

    with self._client as open_client:
        hive_table = self._get_hive_table(open_client, database_name, table_name)

    return self._convert_hive_into_iceberg(hive_table)

register_table(identifier, metadata_location)

Register a new table using existing metadata.

Parameters:

Name Type Description Default
identifier Union[str, Identifier]

Table identifier for the table

required
metadata_location str

The location to the metadata

required

Returns:

Name Type Description
Table Table

The newly registered table

Raises:

Type Description
TableAlreadyExistsError

If the table already exists

Source code in pyiceberg/catalog/hive.py
def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
    """Register a new table using existing metadata.

    Args:
        identifier Union[str, Identifier]: Table identifier for the table
        metadata_location str: The location to the metadata

    Returns:
        Table: The newly registered table

    Raises:
        TableAlreadyExistsError: If the table already exists
    """
    raise NotImplementedError

rename_table(from_identifier, to_identifier)

Rename a fully classified table name.

Parameters:

Name Type Description Default
from_identifier Union[str, Identifier]

Existing table identifier.

required
to_identifier Union[str, Identifier]

New table identifier.

required

Returns:

Name Type Description
Table Table

the updated table instance with its metadata.

Raises:

Type Description
ValueError

When from table identifier is invalid.

NoSuchTableError

When a table with the name does not exist.

NoSuchNamespaceError

When the destination namespace doesn't exist.

Source code in pyiceberg/catalog/hive.py
def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: Union[str, Identifier]) -> Table:
    """Rename a fully classified table name.

    Args:
        from_identifier: Existing table identifier.
        to_identifier: New table identifier.

    Returns:
        Table: the updated table instance with its metadata.

    Raises:
        ValueError: When from table identifier is invalid.
        NoSuchTableError: When a table with the name does not exist.
        NoSuchNamespaceError: When the destination namespace doesn't exist.
    """
    from_identifier_tuple = self._identifier_to_tuple_without_catalog(from_identifier)
    from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
    to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
    try:
        with self._client as open_client:
            tbl = open_client.get_table(dbname=from_database_name, tbl_name=from_table_name)
            tbl.dbName = to_database_name
            tbl.tableName = to_table_name
            open_client.alter_table(dbname=from_database_name, tbl_name=from_table_name, new_tbl=tbl)
    except NoSuchObjectException as e:
        raise NoSuchTableError(f"Table does not exist: {from_table_name}") from e
    except InvalidOperationException as e:
        raise NoSuchNamespaceError(f"Database does not exists: {to_database_name}") from e
    return self.load_table(to_identifier)

update_namespace_properties(namespace, removals=None, updates=EMPTY_DICT)

Remove provided property keys and update properties for a namespace.

Parameters:

Name Type Description Default
namespace Union[str, Identifier]

Namespace identifier.

required
removals Optional[Set[str]]

Set of property keys that need to be removed. Optional Argument.

None
updates Properties

Properties to be updated for the given namespace.

EMPTY_DICT

Raises:

Type Description
NoSuchNamespaceError

If a namespace with the given name does not exist

ValueError

If removals and updates have overlapping keys.

Source code in pyiceberg/catalog/hive.py
def update_namespace_properties(
    self, namespace: Union[str, Identifier], removals: Optional[Set[str]] = None, updates: Properties = EMPTY_DICT
) -> PropertiesUpdateSummary:
    """Remove provided property keys and update properties for a namespace.

    Args:
        namespace: Namespace identifier.
        removals: Set of property keys that need to be removed. Optional Argument.
        updates: Properties to be updated for the given namespace.

    Raises:
        NoSuchNamespaceError: If a namespace with the given name does not exist
        ValueError: If removals and updates have overlapping keys.
    """
    self._check_for_overlap(updates=updates, removals=removals)
    database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)
    with self._client as open_client:
        try:
            database = open_client.get_database(database_name)
            parameters = database.parameters
        except NoSuchObjectException as e:
            raise NoSuchNamespaceError(f"Database does not exists: {database_name}") from e

        removed: Set[str] = set()
        updated: Set[str] = set()

        if removals:
            for key in removals:
                if key in parameters:
                    parameters[key] = None
                    removed.add(key)
        if updates:
            for key, value in updates.items():
                parameters[key] = value
                updated.add(key)

        open_client.alter_database(database_name, _annotate_namespace(database, parameters))

    expected_to_change = (removals or set()).difference(removed)

    return PropertiesUpdateSummary(removed=list(removed or []), updated=list(updated or []), missing=list(expected_to_change))