ebyhr commented on code in PR #3474:
URL: https://github.com/apache/iceberg-python/pull/3474#discussion_r3384501000
##########
pyiceberg/table/puffin.py:
##########
@@ -114,3 +147,105 @@ def __init__(self, puffin: bytes) -> None:
def to_vector(self) -> dict[str, "pa.ChunkedArray"]:
return {path: _bitmaps_to_chunked_array(bitmaps) for path, bitmaps in
self._deletion_vectors.items()}
+
+
+class PuffinWriter:
+ _blobs: list[PuffinBlobMetadata]
+ _blob_payloads: list[bytes]
+ _created_by: str | None
+
+ def __init__(self, created_by: str | None = None) -> None:
+ self._blobs = []
+ self._blob_payloads = []
+ self._created_by = created_by
+
+ def set_blob(
+ self,
+ positions: Iterable[int],
+ referenced_data_file: str,
+ ) -> None:
+ # We only support one blob at the moment
+ self._blobs = []
+ self._blob_payloads = []
+
+ # 1. Create bitmaps from positions
Review Comment:
nit: I would avoid using number prefixes. When we want to add a new
operation, we need to adjust the subsequent numbers.
##########
tests/integration/test_puffin_spark_interop.py:
##########
@@ -0,0 +1,93 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+import pytest
+from pyspark.sql import SparkSession
+
+from pyiceberg.catalog.rest import RestCatalog
+from pyiceberg.manifest import ManifestContent
+from pyiceberg.table.puffin import PuffinFile
+
+
+def run_spark_commands(spark: SparkSession, sqls: list[str]) -> None:
+ for sql in sqls:
+ spark.sql(sql)
+
+
[email protected]
+def test_read_spark_written_puffin_dv(spark: SparkSession, session_catalog:
RestCatalog) -> None:
+ """Verify pyiceberg can read Puffin DVs written by Spark."""
+ identifier = "default.spark_puffin_format_test"
Review Comment:
This PR introduces support for write operations, so we're interested in
verifying that Spark can read Puffin files written by PyIceberg. There are no
requested changes for now. I suppose this PR is a preparatory change, and we'll
need another PR to use it during the write operations.
##########
pyiceberg/table/puffin.py:
##########
@@ -114,3 +147,105 @@ def __init__(self, puffin: bytes) -> None:
def to_vector(self) -> dict[str, "pa.ChunkedArray"]:
return {path: _bitmaps_to_chunked_array(bitmaps) for path, bitmaps in
self._deletion_vectors.items()}
+
+
+class PuffinWriter:
+ _blobs: list[PuffinBlobMetadata]
+ _blob_payloads: list[bytes]
+ _created_by: str | None
Review Comment:
Could you please set the default value for the `_created_by` field using
`PyIceberg version {version}`? You can obtain the version by using
`importlib.metadata.version`.
##########
tests/integration/test_puffin_spark_interop.py:
##########
@@ -0,0 +1,93 @@
+# Licensed to the Apache Software Foundation (ASF) under one
Review Comment:
This test passes without the changes made in this PR. Could you please
extract a PR that adding this test?
##########
pyiceberg/table/puffin.py:
##########
@@ -114,3 +147,105 @@ def __init__(self, puffin: bytes) -> None:
def to_vector(self) -> dict[str, "pa.ChunkedArray"]:
return {path: _bitmaps_to_chunked_array(bitmaps) for path, bitmaps in
self._deletion_vectors.items()}
+
+
+class PuffinWriter:
+ _blobs: list[PuffinBlobMetadata]
+ _blob_payloads: list[bytes]
+ _created_by: str | None
+
+ def __init__(self, created_by: str | None = None) -> None:
+ self._blobs = []
+ self._blob_payloads = []
+ self._created_by = created_by
+
+ def set_blob(
+ self,
+ positions: Iterable[int],
+ referenced_data_file: str,
+ ) -> None:
+ # We only support one blob at the moment
+ self._blobs = []
+ self._blob_payloads = []
+
+ # 1. Create bitmaps from positions
+ bitmaps: dict[int, BitMap] = {}
+ for pos in positions:
+ key = pos >> 32
+ low_bits = pos & 0xFFFFFFFF
+ if key not in bitmaps:
+ bitmaps[key] = BitMap()
+ bitmaps[key].add(low_bits)
+
+ # Calculate the cardinality from the bitmaps
+ cardinality = sum(len(bm) for bm in bitmaps.values())
Review Comment:
nit: A comment for a simple single line seems excessive. It's evident when
we read the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
