Skip to content

Update table metadata throughout transaction #471

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 8 commits into from
Feb 29, 2024
Merged

Update table metadata throughout transaction #471

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
07ab4d2
Update table metadata throughout transaction
Fokko Feb 25, 2024
17bf189
Fix integration tests
Fokko Feb 25, 2024
564dbb3
Merge branch 'main' of github.com:apache/iceberg-python into fd-stack…
Fokko Feb 25, 2024
c75ceae
Thanks Honah!
Fokko Feb 27, 2024
54a2897
Merge branch 'main' of github.com:apache/iceberg-python into fd-stack…
Fokko Feb 28, 2024
5c38c2a
Include the partition evolution
Fokko Feb 28, 2024
161db05
Merge branch 'main' of github.com:apache/iceberg-python into fd-stack…
Fokko Feb 29, 2024
4f95939
Cleanup
Fokko Feb 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 11 additions & 10 deletions pyiceberg/io/pyarrow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@
visit_with_partner,
)
from pyiceberg.table import PropertyUtil, TableProperties, WriteTask
from pyiceberg.table.metadata import TableMetadata
from pyiceberg.table.name_mapping import NameMapping
from pyiceberg.transforms import TruncateTransform
from pyiceberg.typedef import EMPTY_DICT, Properties, Record
Expand Down Expand Up @@ -1720,7 +1721,7 @@ def fill_parquet_file_metadata(
data_file.split_offsets = split_offsets


def write_file(table: Table, tasks: Iterator[WriteTask], file_schema: Optional[Schema] = None) -> Iterator[DataFile]:
def write_file(io: FileIO, table_metadata: TableMetadata, tasks: Iterator[WriteTask]) -> Iterator[DataFile]:
task = next(tasks)

try:
Expand All @@ -1730,15 +1731,15 @@ def write_file(table: Table, tasks: Iterator[WriteTask], file_schema: Optional[S
except StopIteration:
pass

parquet_writer_kwargs = _get_parquet_writer_kwargs(table.properties)
parquet_writer_kwargs = _get_parquet_writer_kwargs(table_metadata.properties)

file_path = f'{table.location()}/data/{task.generate_data_file_filename("parquet")}'
file_schema = file_schema or table.schema()
arrow_file_schema = schema_to_pyarrow(file_schema)
file_path = f'{table_metadata.location}/data/{task.generate_data_file_filename("parquet")}'
schema = table_metadata.schema()
arrow_file_schema = schema_to_pyarrow(schema)

fo = table.io.new_output(file_path)
fo = io.new_output(file_path)
row_group_size = PropertyUtil.property_as_int(
properties=table.properties,
properties=table_metadata.properties,
property_name=TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES,
default=TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT,
)
Expand All @@ -1757,16 +1758,16 @@ def write_file(table: Table, tasks: Iterator[WriteTask], file_schema: Optional[S
# sort_order_id=task.sort_order_id,
sort_order_id=None,
# Just copy these from the table for now
spec_id=table.spec().spec_id,
spec_id=table_metadata.default_spec_id,
equality_ids=None,
key_metadata=None,
)

fill_parquet_file_metadata(
data_file=data_file,
parquet_metadata=writer.writer.metadata,
stats_columns=compute_statistics_plan(file_schema, table.properties),
parquet_column_mapping=parquet_path_to_id_mapping(file_schema),
stats_columns=compute_statistics_plan(schema, table_metadata.properties),
parquet_column_mapping=parquet_path_to_id_mapping(schema),
)
return iter([data_file])

Expand Down
Loading