Merge branch 'main' of github.com:apache/iceberg-python into fd-add-ability-to-delete-full-data-files

Author: Fokko

mkdocs/requirements.txt

@@ -17,12 +17,12 @@
 
 mkdocs==1.6.0
 griffe==0.44.0
-jinja2==3.1.3
-mkdocstrings==0.25.0
+jinja2==3.1.4
+mkdocstrings==0.25.1
 mkdocstrings-python==1.10.0
 mkdocs-literate-nav==0.6.1
 mkdocs-autorefs==1.0.1
 mkdocs-gen-files==0.5.0
-mkdocs-material==9.5.20
+mkdocs-material==9.5.21
 mkdocs-material-extensions==1.3.1
-mkdocs-section-index==0.3.8
+mkdocs-section-index==0.3.9

poetry.lock

@@ -779,7 +779,7 @@ def _get_updated_props_and_update_summary(
     def _resolve_table_location(self, location: Optional[str], database_name: str, table_name: str) -> str:
         if not location:
             return self._get_default_warehouse_location(database_name, table_name)
-        return location
+        return location.rstrip("/")
 
     def _get_default_warehouse_location(self, database_name: str, table_name: str) -> str:
         database_properties = self.load_namespace_properties(database_name)

pyiceberg/catalog/__init__.py

@@ -519,6 +519,8 @@ def _create_table(
         fresh_sort_order = assign_fresh_sort_order_ids(sort_order, iceberg_schema, fresh_schema)
 
         namespace_and_table = self._split_identifier_for_path(identifier)
+        if location:
+            location = location.rstrip("/")
         request = CreateTableRequest(
             name=namespace_and_table["table"],
             location=location,

pyiceberg/catalog/rest.py

@@ -731,6 +731,16 @@ def _(obj: pa.MapType, visitor: PyArrowSchemaVisitor[T]) -> T:
     return visitor.map(obj, key_result, value_result)
 
 
+@visit_pyarrow.register(pa.DictionaryType)
+def _(obj: pa.DictionaryType, visitor: PyArrowSchemaVisitor[T]) -> T:
+    # Parquet has no dictionary type. dictionary-encoding is handled
+    # as an encoding detail, not as a separate type.
+    # We will follow this approach in determining the Iceberg Type,
+    # as we only support parquet in PyIceberg for now.
+    logger.warning(f"Iceberg does not have a dictionary type. {type(obj)} will be inferred as {obj.value_type} on read.")
+    return visit_pyarrow(obj.value_type, visitor)
+
+
 @visit_pyarrow.register(pa.DataType)
 def _(obj: pa.DataType, visitor: PyArrowSchemaVisitor[T]) -> T:
     if pa.types.is_nested(obj):

pyiceberg/io/pyarrow.py

@@ -539,7 +539,7 @@ def delete(self, delete_filter: Union[str, BooleanExpression], snapshot_properti
         if not delete_snapshot.files_affected and not delete_snapshot.rewrites_needed:
             warnings.warn("Delete operation did not match any records")
 
-    def add_files(self, file_paths: List[str]) -> None:
+    def add_files(self, file_paths: List[str], snapshot_properties: Dict[str, str] = EMPTY_DICT) -> None:
         """
         Shorthand API for adding files as data files to the table transaction.
 
@@ -551,7 +551,7 @@ def add_files(self, file_paths: List[str]) -> None:
         """
         if self._table.name_mapping() is None:
             self.set_properties(**{TableProperties.DEFAULT_NAME_MAPPING: self._table.schema().name_mapping.model_dump_json()})
-        with self.update_snapshot().fast_append() as update_snapshot:
+        with self.update_snapshot(snapshot_properties=snapshot_properties).fast_append() as update_snapshot:
             data_files = _parquet_files_to_data_files(
                 table_metadata=self._table.metadata, file_paths=file_paths, io=self._table.io
             )
@@ -1443,6 +1443,7 @@ def overwrite(
         with self.transaction() as tx:
             tx.overwrite(df=df, overwrite_filter=overwrite_filter, snapshot_properties=snapshot_properties)
 
+<<<<<<< HEAD
     def delete(
         self, delete_filter: Union[BooleanExpression, str] = ALWAYS_TRUE, snapshot_properties: Dict[str, str] = EMPTY_DICT
     ) -> None:
@@ -1457,6 +1458,9 @@ def delete(
             tx.delete(delete_filter=delete_filter, snapshot_properties=snapshot_properties)
 
     def add_files(self, file_paths: List[str]) -> None:
+=======
+    def add_files(self, file_paths: List[str], snapshot_properties: Dict[str, str] = EMPTY_DICT) -> None:
+>>>>>>> aa361d1485f4a914bc0bbc2e574becaec9a773ac
         """
         Shorthand API for adding files as data files to the table.
 
@@ -1467,7 +1471,7 @@ def add_files(self, file_paths: List[str]) -> None:
             FileNotFoundError: If the file does not exist.
         """
         with self.transaction() as tx:
-            tx.add_files(file_paths=file_paths)
+            tx.add_files(file_paths=file_paths, snapshot_properties=snapshot_properties)
 
     def update_spec(self, case_sensitive: bool = True) -> UpdateSpec:
         return UpdateSpec(Transaction(self, autocommit=True), case_sensitive=case_sensitive)

pyiceberg/table/__init__.py

@@ -85,7 +85,7 @@ moto = { version = "^5.0.2", extras = ["server"] }
 typing-extensions = "4.11.0"
 pytest-mock = "3.14.0"
 pyspark = "3.5.1"
-cython = "3.0.8"
+cython = "3.0.10"
 deptry = ">=0.14,<0.17"
 docutils = "!=0.21"
 

pyproject.toml

@@ -262,3 +262,9 @@ def test_update_namespace_properties(test_catalog: Catalog, database_name: str)
         else:
             assert k in update_report.removed
     assert "updated test description" == test_catalog.load_namespace_properties(database_name)["comment"]
+
+
+def test_table_exists(test_catalog: Catalog, table_schema_nested: Schema, database_name: str, table_name: str) -> None:
+    test_catalog.create_namespace(database_name)
+    test_catalog.create_table((database_name, table_name), table_schema_nested)
+    assert test_catalog.table_exists((database_name, table_name)) is True

tests/catalog/integration_test_dynamodb.py

@@ -557,3 +557,9 @@ def test_create_table_transaction(
             ]
         },
     ]
+
+
+def test_table_exists(test_catalog: Catalog, table_schema_nested: Schema, table_name: str, database_name: str) -> None:
+    test_catalog.create_namespace(database_name)
+    test_catalog.create_table((database_name, table_name), table_schema_nested)
+    assert test_catalog.table_exists((database_name, table_name)) is True

tests/catalog/integration_test_glue.py

@@ -105,6 +105,7 @@ def create_table(
 
             if not location:
                 location = f'{self._warehouse_location}/{"/".join(identifier)}'
+            location = location.rstrip("/")
 
             metadata_location = self._get_metadata_location(location=location)
             metadata = new_table_metadata(
@@ -353,6 +354,19 @@ def test_create_table_location_override(catalog: InMemoryCatalog) -> None:
     assert table.location() == new_location
 
 
+def test_create_table_removes_trailing_slash_from_location(catalog: InMemoryCatalog) -> None:
+    new_location = f"{catalog._warehouse_location}/new_location"
+    table = catalog.create_table(
+        identifier=TEST_TABLE_IDENTIFIER,
+        schema=TEST_TABLE_SCHEMA,
+        location=f"{new_location}/",
+        partition_spec=TEST_TABLE_PARTITION_SPEC,
+        properties=TEST_TABLE_PROPERTIES,
+    )
+    assert catalog.load_table(TEST_TABLE_IDENTIFIER) == table
+    assert table.location() == new_location
+
+
 @pytest.mark.parametrize(
     "schema,expected",
     [