PyArrow: Keep the row count when a scan selects no columns

pyiceberg/io/pyarrow.py

@@ -1888,6 +1888,10 @@ def to_table(self, tasks: Iterable[FileScanTask]) -> pa.Table:
         arrow_schema = schema_to_pyarrow(self._projected_schema, include_field_ids=False)
 
         batches = self.to_record_batches(tasks)
+        if not self._projected_schema.fields:
+            # pa.concat_tables drops the rows of tables without columns
+            return pa.Table.from_batches(batches, schema=arrow_schema)
+
         try:
             first_batch = next(batches)
         except StopIteration:
@@ -1990,6 +1994,10 @@ def _to_requested_schema(
     allow_timestamp_tz_mismatch: bool = False,
     format_model: FileFormatModel | None = None,
 ) -> pa.RecordBatch:
+    if not requested_schema.fields:
+        # A struct array without fields has length 0, so keep the rows of the batch instead
+        return batch.select([])
+
     # We could reuse some of these visitors
     struct_array = visit_with_partner(
         requested_schema,

tests/io/test_pyarrow.py

@@ -1336,6 +1336,20 @@ def test_projection_concat_files(schema_int: Schema, file_int: str) -> None:
     assert repr(result_table.schema) == "id: int32"
 
 
+def test_projection_no_columns_keeps_row_count(schema_int: Schema, file_int: str) -> None:
+    result_table = project(Schema(), [file_int, file_int], table_schema=schema_int)
+
+    assert result_table.num_columns == 0
+    assert result_table.num_rows == 6
+
+
+def test_projection_no_columns_with_filter(schema_int: Schema, file_int: str) -> None:
+    result_table = project(Schema(), [file_int], expr=GreaterThan("id", 0), table_schema=schema_int)
+
+    assert result_table.num_columns == 0
+    assert result_table.num_rows == 2
+
+
 @pytest.mark.parametrize("with_field_ids", [False, True], ids=["name-mapping", "field-ids"])
 def test_identity_transform_column_projection(tmp_path: str, catalog: InMemoryCatalog, with_field_ids: bool) -> None:
     # Test by adding a non-partitioned data file to a partitioned table, verifying partition value