static overwrite full table with tests

Author: jqin61

pyiceberg/table/__init__.py

@@ -970,9 +970,6 @@ def overwrite(self, df: pa.Table, overwrite_filter: BooleanExpression = ALWAYS_T
         if overwrite_filter != AlwaysTrue():
             raise NotImplementedError("Cannot overwrite a subset of a table")
 
-        if len(self.spec().fields) > 0:
-            raise ValueError("Cannot write to partitioned tables")
-
         merge = _MergingSnapshotProducer(
             operation=Operation.OVERWRITE if self.current_snapshot() is not None else Operation.APPEND,
             table=self,

tests/integration/test_partitioned_writes.py

@@ -437,6 +437,7 @@ def test_summaries_with_null(spark: SparkSession, session_catalog: Catalog, arro
 
     tbl.append(arrow_table_with_null)
     tbl.append(arrow_table_with_null)
+    tbl.overwrite(arrow_table_with_null)
 
     rows = spark.sql(
         f"""
@@ -447,7 +448,7 @@ def test_summaries_with_null(spark: SparkSession, session_catalog: Catalog, arro
     ).collect()
 
     operations = [row.operation for row in rows]
-    assert operations == ['append', 'append']
+    assert operations == ['append', 'append', 'overwrite']
 
     summaries = [row.summary for row in rows]
 
@@ -474,6 +475,20 @@ def test_summaries_with_null(spark: SparkSession, session_catalog: Catalog, arro
         'total-position-deletes': '0',
         'total-records': '6',
     }
+    assert summaries[2] == {
+        'removed-files-size': '28942',
+        'added-data-files': '3',
+        'total-equality-deletes': '0',
+        'added-records': '3',
+        'total-position-deletes': '0',
+        'deleted-data-files': '6',
+        'added-files-size': '14471',
+        'total-delete-files': '0',
+        'total-files-size': '14471',
+        'deleted-records': '6',
+        'total-records': '3',
+        'total-data-files': '3',
+    }
 
 
 @pytest.mark.integration
@@ -495,6 +510,7 @@ def test_data_files_with_table_partitioned_with_null(
 
     tbl.append(arrow_table_with_null)
     tbl.append(arrow_table_with_null)
+    tbl.overwrite(arrow_table_with_null)
 
     # added_data_files_count, existing_data_files_count, deleted_data_files_count
     rows = spark.sql(
@@ -504,13 +520,9 @@ def test_data_files_with_table_partitioned_with_null(
     """
     ).collect()
 
-    assert [row.added_data_files_count for row in rows] == [3, 3, 3]
-    assert [row.existing_data_files_count for row in rows] == [
-        0,
-        0,
-        0,
-    ]
-    assert [row.deleted_data_files_count for row in rows] == [0, 0, 0]
+    assert [row.added_data_files_count for row in rows] == [3, 3, 3, 3, 0]
+    assert [row.existing_data_files_count for row in rows] == [0, 0, 0, 0, 0]
+    assert [row.deleted_data_files_count for row in rows] == [0, 0, 0, 0, 6]
 
 
 @pytest.mark.integration
@@ -529,5 +541,8 @@ def test_invalid_arguments(spark: SparkSession, session_catalog: Catalog, arrow_
         properties={'format-version': '1'},
     )
 
+    with pytest.raises(ValueError, match="Expected PyArrow table, got: not a df"):
+        tbl.overwrite("not a df")
+
     with pytest.raises(ValueError, match="Expected PyArrow table, got: not a df"):
         tbl.append("not a df")