Arrow: Allow missing field-ids from Schema#183

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

Closed
Fokko wants to merge 4 commits into from
Closed

Arrow: Allow missing field-ids from Schema #183

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
405d36c
Arrow: Allow missing field-ids from Schema
Fokko Dec 5, 2023
27017cf
Thanks Honah!
Fokko Dec 7, 2023
2657fe1
Update pyiceberg/io/pyarrow.py
Fokko Dec 9, 2023
447d22f
lint
Fokko Dec 9, 2023
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
67 changes: 49 additions & 18 deletions pyiceberg/io/pyarrow.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -25,9 +25,11 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
import warnings
from abc import ABC, abstractmethod
from concurrent.futures import Future
from dataclasses import dataclass
Expand DownExpand Up@@ -110,6 +112,7 @@
Schema,
SchemaVisitorPerPrimitiveType,
SchemaWithPartnerVisitor,
assign_fresh_schema_ids,
pre_order_visit,
promote,
prune_columns,
Expand DownExpand Up@@ -616,7 +619,12 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:

def pyarrow_to_schema(schema: pa.Schema) -> Schema:
visitor = _ConvertToIceberg()
return visit_pyarrow(schema, visitor)
schema = visit_pyarrow(schema, visitor)

if visitor.missing_id_metadata:
return assign_fresh_schema_ids(schema)
else:
return schema


@singledispatch
Expand DownExpand Up@@ -713,28 +721,51 @@ def primitive(self, primitive: pa.DataType) -> Optional[T]:
"""Visit a primitive type."""


def _get_field_id(field: pa.Field) -> Optional[int]:
for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
if field_id_str := field.metadata.get(pyarrow_field_id_key):
return int(field_id_str.decode())
return None
class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
counter: itertools.count[int]
missing_id_metadata: Optional[bool]

def __init__(self) -> None:
self.counter = itertools.count(1)
self.missing_id_metadata = None

def _get_field_id(self, field: pa.Field) -> int:
field_id: Optional[int] = None

for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
if field.metadata and (field_id_str := field.metadata.get(pyarrow_field_id_key)):
field_id = int(field_id_str.decode())

if field_id is None:
if self.missing_id_metadata is None:
warnings.warn(
"Missing field-IDs will be auto-assigned, possibly leading to inconsistencies between the file schema and the schema stored in table metadata."
)
field_id = next(self.counter)
missing_is_metadata = True
else:
missing_is_metadata = False

def _get_field_doc(field: pa.Field) -> Optional[str]:
for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
if doc_str := field.metadata.get(pyarrow_doc_key):
return doc_str.decode()
return None
if self.missing_id_metadata is not None and self.missing_id_metadata != missing_is_metadata:
raise ValueError("Parquet file contains partial field-ids")
else:
self.missing_id_metadata = missing_is_metadata

return field_id

def _get_field_doc(self, field: pa.Field) -> Optional[str]:
for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
if field.metadata and (doc_str := field.metadata.get(pyarrow_doc_key)):
return doc_str.decode()
return None

class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
def _convert_fields(self, arrow_fields: Iterable[pa.Field], field_results: List[Optional[IcebergType]]) -> List[NestedField]:
fields = []
for i, field in enumerate(arrow_fields):
field_id = _get_field_id(field)
field_doc = _get_field_doc(field)
field_id = self._get_field_id(field)
field_doc = self._get_field_doc(field)
field_type = field_results[i]
if field_type is not None and field_id is not None:
if field_type is not None:
fields.append(NestedField(field_id, field.name, field_type, required=not field.nullable, doc=field_doc))
return fields

Expand All@@ -746,7 +777,7 @@ def struct(self, struct: pa.StructType, field_results: List[Optional[IcebergType

def list(self, list_type: pa.ListType, element_result: Optional[IcebergType]) -> Optional[IcebergType]:
element_field = list_type.value_field
element_id = _get_field_id(element_field)
element_id = self._get_field_id(element_field)
if element_result is not None and element_id is not None:
return ListType(element_id, element_result, element_required=not element_field.nullable)
return None
Expand All@@ -755,9 +786,9 @@ def map(
self, map_type: pa.MapType, key_result: Optional[IcebergType], value_result: Optional[IcebergType]
) -> Optional[IcebergType]:
key_field = map_type.key_field
key_id = _get_field_id(key_field)
key_id = self._get_field_id(key_field)
value_field = map_type.item_field
value_id = _get_field_id(value_field)
value_id = self._get_field_id(value_field)
if key_result is not None and value_result is not None and key_id is not None and value_id is not None:
return MapType(key_id, key_result, value_id, value_result, value_required=not value_field.nullable)
return None
Expand Down
30 changes: 30 additions & 0 deletions tests/io/test_pyarrow_visitor.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
# pylint: disable=protected-access,unused-argument,redefined-outer-name
import re
from unittest.mock import Mock, patch

import pyarrow as pa
import pytest
Expand DownExpand Up@@ -269,3 +270,32 @@ def test_round_schema_conversion_nested(table_schema_nested: Schema) -> None:
15: person: optional struct<16: name: optional string, 17: age: required int>
}"""
assert actual == expected


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_ids(warn: Mock) -> None:
schema = pa.schema([pa.field('some_int', pa.int32(), nullable=True), pa.field('some_string', pa.string(), nullable=False)])
actual = pyarrow_to_schema(schema)

expected = Schema(
NestedField(field_id=1, name="some_int", field_type=IntegerType(), required=False),
NestedField(field_id=2, name="some_string", field_type=StringType(), required=True),
)

assert actual == expected
assert warn.called


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_id(warn: Mock) -> None:
schema = pa.schema(
[
pa.field('some_int', pa.int32(), nullable=True),
pa.field('some_string', pa.string(), nullable=False, metadata={b"field_id": "22"}),
]
)

with pytest.raises(ValueError) as exc_info:
_ = pyarrow_to_schema(schema)
assert "Parquet file contains partial field-ids" in str(exc_info.value)
assert warn.called
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all 
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
Arrow: Allow missing field-ids from Schema#183

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

 Closed

Fokko
wants to merge
4
commits into

from

 Closed


Arrow: Allow missing field-ids from Schema
#183
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
405d36c

Arrow: Allow missing field-ids from Schema


Fokko Dec 5, 2023
27017cf

Thanks Honah!


Fokko Dec 7, 2023
2657fe1

Update pyiceberg/io/pyarrow.py


Fokko Dec 9, 2023
447d22f

lint


Fokko Dec 9, 2023

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


67 changes: 49 additions & 18 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,9 +25,11 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
import warnings
from abc import ABC, abstractmethod
from concurrent.futures import Future
from dataclasses import dataclass
Expand DownExpand Up@@ -110,6 +112,7 @@
 Schema,
 SchemaVisitorPerPrimitiveType,
 SchemaWithPartnerVisitor,
 assign_fresh_schema_ids,
 pre_order_visit,
 promote,
 prune_columns,
Expand DownExpand Up@@ -616,7 +619,12 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:

def pyarrow_to_schema(schema: pa.Schema) -> Schema:
 visitor = _ConvertToIceberg()
 return visit_pyarrow(schema, visitor)
 schema = visit_pyarrow(schema, visitor)

 if visitor.missing_id_metadata:
 return assign_fresh_schema_ids(schema)
 else:
 return schema


@singledispatch
Expand DownExpand Up@@ -713,28 +721,51 @@ def primitive(self, primitive: pa.DataType) -> Optional[T]:
 """Visit a primitive type."""


def _get_field_id(field: pa.Field) -> Optional[int]:
 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field_id_str := field.metadata.get(pyarrow_field_id_key):
 return int(field_id_str.decode())
 return None
class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 counter: itertools.count[int]
 missing_id_metadata: Optional[bool]

 def __init__(self) -> None:
 self.counter = itertools.count(1)
 self.missing_id_metadata = None

 def _get_field_id(self, field: pa.Field) -> int:
 field_id: Optional[int] = None

 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field.metadata and (field_id_str := field.metadata.get(pyarrow_field_id_key)):
 field_id = int(field_id_str.decode())

 if field_id is None:
 if self.missing_id_metadata is None:
 warnings.warn(
 "Missing field-IDs will be auto-assigned, possibly leading to inconsistencies between the file schema and the schema stored in table metadata."
 )
 field_id = next(self.counter)
 missing_is_metadata = True
 else:
 missing_is_metadata = False

def _get_field_doc(field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if doc_str := field.metadata.get(pyarrow_doc_key):
 return doc_str.decode()
 return None
 if self.missing_id_metadata is not None and self.missing_id_metadata != missing_is_metadata:
 raise ValueError("Parquet file contains partial field-ids")
 else:
 self.missing_id_metadata = missing_is_metadata

 return field_id

 def _get_field_doc(self, field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if field.metadata and (doc_str := field.metadata.get(pyarrow_doc_key)):
 return doc_str.decode()
 return None

class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 def _convert_fields(self, arrow_fields: Iterable[pa.Field], field_results: List[Optional[IcebergType]]) -> List[NestedField]:
 fields = []
 for i, field in enumerate(arrow_fields):
 field_id = _get_field_id(field)
 field_doc = _get_field_doc(field)
 field_id = self._get_field_id(field)
 field_doc = self._get_field_doc(field)
 field_type = field_results[i]
 if field_type is not None and field_id is not None:
 if field_type is not None:
 fields.append(NestedField(field_id, field.name, field_type, required=not field.nullable, doc=field_doc))
 return fields

Expand All@@ -746,7 +777,7 @@ def struct(self, struct: pa.StructType, field_results: List[Optional[IcebergType

 def list(self, list_type: pa.ListType, element_result: Optional[IcebergType]) -> Optional[IcebergType]:
 element_field = list_type.value_field
 element_id = _get_field_id(element_field)
 element_id = self._get_field_id(element_field)
 if element_result is not None and element_id is not None:
 return ListType(element_id, element_result, element_required=not element_field.nullable)
 return None
Expand All@@ -755,9 +786,9 @@ def map(
 self, map_type: pa.MapType, key_result: Optional[IcebergType], value_result: Optional[IcebergType]
 ) -> Optional[IcebergType]:
 key_field = map_type.key_field
 key_id = _get_field_id(key_field)
 key_id = self._get_field_id(key_field)
 value_field = map_type.item_field
 value_id = _get_field_id(value_field)
 value_id = self._get_field_id(value_field)
 if key_result is not None and value_result is not None and key_id is not None and value_id is not None:
 return MapType(key_id, key_result, value_id, value_result, value_required=not value_field.nullable)
 return None
Expand Down

30 changes: 30 additions & 0 deletions
tests/io/test_pyarrow_visitor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
# pylint: disable=protected-access,unused-argument,redefined-outer-name
import re
from unittest.mock import Mock, patch

import pyarrow as pa
import pytest
Expand DownExpand Up@@ -269,3 +270,32 @@ def test_round_schema_conversion_nested(table_schema_nested: Schema) -> None:
 15: person: optional struct<16: name: optional string, 17: age: required int>
}"""
 assert actual == expected


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_ids(warn: Mock) -> None:
 schema = pa.schema([pa.field('some_int', pa.int32(), nullable=True), pa.field('some_string', pa.string(), nullable=False)])
 actual = pyarrow_to_schema(schema)

 expected = Schema(
 NestedField(field_id=1, name="some_int", field_type=IntegerType(), required=False),
 NestedField(field_id=2, name="some_string", field_type=StringType(), required=True),
 )

 assert actual == expected
 assert warn.called


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_id(warn: Mock) -> None:
 schema = pa.schema(
 [
 pa.field('some_int', pa.int32(), nullable=True),
 pa.field('some_string', pa.string(), nullable=False, metadata={b"field_id": "22"}),
 ]
 )

 with pytest.raises(ValueError) as exc_info:
 _ = pyarrow_to_schema(schema)
 assert "Parquet file contains partial field-ids" in str(exc_info.value)
 assert warn.called

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix");
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Arrow: Allow missing field-ids from Schema#183

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

 Closed

Fokko
wants to merge
4
commits into

from

 Closed


Arrow: Allow missing field-ids from Schema
#183
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
405d36c

Arrow: Allow missing field-ids from Schema


Fokko Dec 5, 2023
27017cf

Thanks Honah!


Fokko Dec 7, 2023
2657fe1

Update pyiceberg/io/pyarrow.py


Fokko Dec 9, 2023
447d22f

lint


Fokko Dec 9, 2023

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


67 changes: 49 additions & 18 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,9 +25,11 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
import warnings
from abc import ABC, abstractmethod
from concurrent.futures import Future
from dataclasses import dataclass
Expand DownExpand Up@@ -110,6 +112,7 @@
 Schema,
 SchemaVisitorPerPrimitiveType,
 SchemaWithPartnerVisitor,
 assign_fresh_schema_ids,
 pre_order_visit,
 promote,
 prune_columns,
Expand DownExpand Up@@ -616,7 +619,12 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:

def pyarrow_to_schema(schema: pa.Schema) -> Schema:
 visitor = _ConvertToIceberg()
 return visit_pyarrow(schema, visitor)
 schema = visit_pyarrow(schema, visitor)

 if visitor.missing_id_metadata:
 return assign_fresh_schema_ids(schema)
 else:
 return schema


@singledispatch
Expand DownExpand Up@@ -713,28 +721,51 @@ def primitive(self, primitive: pa.DataType) -> Optional[T]:
 """Visit a primitive type."""


def _get_field_id(field: pa.Field) -> Optional[int]:
 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field_id_str := field.metadata.get(pyarrow_field_id_key):
 return int(field_id_str.decode())
 return None
class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 counter: itertools.count[int]
 missing_id_metadata: Optional[bool]

 def __init__(self) -> None:
 self.counter = itertools.count(1)
 self.missing_id_metadata = None

 def _get_field_id(self, field: pa.Field) -> int:
 field_id: Optional[int] = None

 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field.metadata and (field_id_str := field.metadata.get(pyarrow_field_id_key)):
 field_id = int(field_id_str.decode())

 if field_id is None:
 if self.missing_id_metadata is None:
 warnings.warn(
 "Missing field-IDs will be auto-assigned, possibly leading to inconsistencies between the file schema and the schema stored in table metadata."
 )
 field_id = next(self.counter)
 missing_is_metadata = True
 else:
 missing_is_metadata = False

def _get_field_doc(field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if doc_str := field.metadata.get(pyarrow_doc_key):
 return doc_str.decode()
 return None
 if self.missing_id_metadata is not None and self.missing_id_metadata != missing_is_metadata:
 raise ValueError("Parquet file contains partial field-ids")
 else:
 self.missing_id_metadata = missing_is_metadata

 return field_id

 def _get_field_doc(self, field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if field.metadata and (doc_str := field.metadata.get(pyarrow_doc_key)):
 return doc_str.decode()
 return None

class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 def _convert_fields(self, arrow_fields: Iterable[pa.Field], field_results: List[Optional[IcebergType]]) -> List[NestedField]:
 fields = []
 for i, field in enumerate(arrow_fields):
 field_id = _get_field_id(field)
 field_doc = _get_field_doc(field)
 field_id = self._get_field_id(field)
 field_doc = self._get_field_doc(field)
 field_type = field_results[i]
 if field_type is not None and field_id is not None:
 if field_type is not None:
 fields.append(NestedField(field_id, field.name, field_type, required=not field.nullable, doc=field_doc))
 return fields

Expand All@@ -746,7 +777,7 @@ def struct(self, struct: pa.StructType, field_results: List[Optional[IcebergType

 def list(self, list_type: pa.ListType, element_result: Optional[IcebergType]) -> Optional[IcebergType]:
 element_field = list_type.value_field
 element_id = _get_field_id(element_field)
 element_id = self._get_field_id(element_field)
 if element_result is not None and element_id is not None:
 return ListType(element_id, element_result, element_required=not element_field.nullable)
 return None
Expand All@@ -755,9 +786,9 @@ def map(
 self, map_type: pa.MapType, key_result: Optional[IcebergType], value_result: Optional[IcebergType]
 ) -> Optional[IcebergType]:
 key_field = map_type.key_field
 key_id = _get_field_id(key_field)
 key_id = self._get_field_id(key_field)
 value_field = map_type.item_field
 value_id = _get_field_id(value_field)
 value_id = self._get_field_id(value_field)
 if key_result is not None and value_result is not None and key_id is not None and value_id is not None:
 return MapType(key_id, key_result, value_id, value_result, value_required=not value_field.nullable)
 return None
Expand Down

30 changes: 30 additions & 0 deletions
tests/io/test_pyarrow_visitor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
# pylint: disable=protected-access,unused-argument,redefined-outer-name
import re
from unittest.mock import Mock, patch

import pyarrow as pa
import pytest
Expand DownExpand Up@@ -269,3 +270,32 @@ def test_round_schema_conversion_nested(table_schema_nested: Schema) -> None:
 15: person: optional struct<16: name: optional string, 17: age: required int>
}"""
 assert actual == expected


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_ids(warn: Mock) -> None:
 schema = pa.schema([pa.field('some_int', pa.int32(), nullable=True), pa.field('some_string', pa.string(), nullable=False)])
 actual = pyarrow_to_schema(schema)

 expected = Schema(
 NestedField(field_id=1, name="some_int", field_type=IntegerType(), required=False),
 NestedField(field_id=2, name="some_string", field_type=StringType(), required=True),
 )

 assert actual == expected
 assert warn.called


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_id(warn: Mock) -> None:
 schema = pa.schema(
 [
 pa.field('some_int', pa.int32(), nullable=True),
 pa.field('some_string', pa.string(), nullable=False, metadata={b"field_id": "22"}),
 ]
 )

 with pytest.raises(ValueError) as exc_info:
 _ = pyarrow_to_schema(schema)
 assert "Parquet file contains partial field-ids" in str(exc_info.value)
 assert warn.called

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms");
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Arrow: Allow missing field-ids from Schema#183

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

 Closed

Fokko
wants to merge
4
commits into

from

 Closed


Arrow: Allow missing field-ids from Schema
#183
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
405d36c

Arrow: Allow missing field-ids from Schema


Fokko Dec 5, 2023
27017cf

Thanks Honah!


Fokko Dec 7, 2023
2657fe1

Update pyiceberg/io/pyarrow.py


Fokko Dec 9, 2023
447d22f

lint


Fokko Dec 9, 2023

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


67 changes: 49 additions & 18 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,9 +25,11 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
import warnings
from abc import ABC, abstractmethod
from concurrent.futures import Future
from dataclasses import dataclass
Expand DownExpand Up@@ -110,6 +112,7 @@
 Schema,
 SchemaVisitorPerPrimitiveType,
 SchemaWithPartnerVisitor,
 assign_fresh_schema_ids,
 pre_order_visit,
 promote,
 prune_columns,
Expand DownExpand Up@@ -616,7 +619,12 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:

def pyarrow_to_schema(schema: pa.Schema) -> Schema:
 visitor = _ConvertToIceberg()
 return visit_pyarrow(schema, visitor)
 schema = visit_pyarrow(schema, visitor)

 if visitor.missing_id_metadata:
 return assign_fresh_schema_ids(schema)
 else:
 return schema


@singledispatch
Expand DownExpand Up@@ -713,28 +721,51 @@ def primitive(self, primitive: pa.DataType) -> Optional[T]:
 """Visit a primitive type."""


def _get_field_id(field: pa.Field) -> Optional[int]:
 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field_id_str := field.metadata.get(pyarrow_field_id_key):
 return int(field_id_str.decode())
 return None
class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 counter: itertools.count[int]
 missing_id_metadata: Optional[bool]

 def __init__(self) -> None:
 self.counter = itertools.count(1)
 self.missing_id_metadata = None

 def _get_field_id(self, field: pa.Field) -> int:
 field_id: Optional[int] = None

 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field.metadata and (field_id_str := field.metadata.get(pyarrow_field_id_key)):
 field_id = int(field_id_str.decode())

 if field_id is None:
 if self.missing_id_metadata is None:
 warnings.warn(
 "Missing field-IDs will be auto-assigned, possibly leading to inconsistencies between the file schema and the schema stored in table metadata."
 )
 field_id = next(self.counter)
 missing_is_metadata = True
 else:
 missing_is_metadata = False

def _get_field_doc(field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if doc_str := field.metadata.get(pyarrow_doc_key):
 return doc_str.decode()
 return None
 if self.missing_id_metadata is not None and self.missing_id_metadata != missing_is_metadata:
 raise ValueError("Parquet file contains partial field-ids")
 else:
 self.missing_id_metadata = missing_is_metadata

 return field_id

 def _get_field_doc(self, field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if field.metadata and (doc_str := field.metadata.get(pyarrow_doc_key)):
 return doc_str.decode()
 return None

class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 def _convert_fields(self, arrow_fields: Iterable[pa.Field], field_results: List[Optional[IcebergType]]) -> List[NestedField]:
 fields = []
 for i, field in enumerate(arrow_fields):
 field_id = _get_field_id(field)
 field_doc = _get_field_doc(field)
 field_id = self._get_field_id(field)
 field_doc = self._get_field_doc(field)
 field_type = field_results[i]
 if field_type is not None and field_id is not None:
 if field_type is not None:
 fields.append(NestedField(field_id, field.name, field_type, required=not field.nullable, doc=field_doc))
 return fields

Expand All@@ -746,7 +777,7 @@ def struct(self, struct: pa.StructType, field_results: List[Optional[IcebergType

 def list(self, list_type: pa.ListType, element_result: Optional[IcebergType]) -> Optional[IcebergType]:
 element_field = list_type.value_field
 element_id = _get_field_id(element_field)
 element_id = self._get_field_id(element_field)
 if element_result is not None and element_id is not None:
 return ListType(element_id, element_result, element_required=not element_field.nullable)
 return None
Expand All@@ -755,9 +786,9 @@ def map(
 self, map_type: pa.MapType, key_result: Optional[IcebergType], value_result: Optional[IcebergType]
 ) -> Optional[IcebergType]:
 key_field = map_type.key_field
 key_id = _get_field_id(key_field)
 key_id = self._get_field_id(key_field)
 value_field = map_type.item_field
 value_id = _get_field_id(value_field)
 value_id = self._get_field_id(value_field)
 if key_result is not None and value_result is not None and key_id is not None and value_id is not None:
 return MapType(key_id, key_result, value_id, value_result, value_required=not value_field.nullable)
 return None
Expand Down

30 changes: 30 additions & 0 deletions
tests/io/test_pyarrow_visitor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
# pylint: disable=protected-access,unused-argument,redefined-outer-name
import re
from unittest.mock import Mock, patch

import pyarrow as pa
import pytest
Expand DownExpand Up@@ -269,3 +270,32 @@ def test_round_schema_conversion_nested(table_schema_nested: Schema) -> None:
 15: person: optional struct<16: name: optional string, 17: age: required int>
}"""
 assert actual == expected


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_ids(warn: Mock) -> None:
 schema = pa.schema([pa.field('some_int', pa.int32(), nullable=True), pa.field('some_string', pa.string(), nullable=False)])
 actual = pyarrow_to_schema(schema)

 expected = Schema(
 NestedField(field_id=1, name="some_int", field_type=IntegerType(), required=False),
 NestedField(field_id=2, name="some_string", field_type=StringType(), required=True),
 )

 assert actual == expected
 assert warn.called


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_id(warn: Mock) -> None:
 schema = pa.schema(
 [
 pa.field('some_int', pa.int32(), nullable=True),
 pa.field('some_string', pa.string(), nullable=False, metadata={b"field_id": "22"}),
 ]
 )

 with pytest.raises(ValueError) as exc_info:
 _ = pyarrow_to_schema(schema)
 assert "Parquet file contains partial field-ids" in str(exc_info.value)
 assert warn.called

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links");
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
Arrow: Allow missing field-ids from Schema#183

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

 Closed

Fokko
wants to merge
4
commits into

from

 Closed


Arrow: Allow missing field-ids from Schema
#183
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
405d36c

Arrow: Allow missing field-ids from Schema


Fokko Dec 5, 2023
27017cf

Thanks Honah!


Fokko Dec 7, 2023
2657fe1

Update pyiceberg/io/pyarrow.py


Fokko Dec 9, 2023
447d22f

lint


Fokko Dec 9, 2023

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


67 changes: 49 additions & 18 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,9 +25,11 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
import warnings
from abc import ABC, abstractmethod
from concurrent.futures import Future
from dataclasses import dataclass
Expand DownExpand Up@@ -110,6 +112,7 @@
 Schema,
 SchemaVisitorPerPrimitiveType,
 SchemaWithPartnerVisitor,
 assign_fresh_schema_ids,
 pre_order_visit,
 promote,
 prune_columns,
Expand DownExpand Up@@ -616,7 +619,12 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:

def pyarrow_to_schema(schema: pa.Schema) -> Schema:
 visitor = _ConvertToIceberg()
 return visit_pyarrow(schema, visitor)
 schema = visit_pyarrow(schema, visitor)

 if visitor.missing_id_metadata:
 return assign_fresh_schema_ids(schema)
 else:
 return schema


@singledispatch
Expand DownExpand Up@@ -713,28 +721,51 @@ def primitive(self, primitive: pa.DataType) -> Optional[T]:
 """Visit a primitive type."""


def _get_field_id(field: pa.Field) -> Optional[int]:
 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field_id_str := field.metadata.get(pyarrow_field_id_key):
 return int(field_id_str.decode())
 return None
class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 counter: itertools.count[int]
 missing_id_metadata: Optional[bool]

 def __init__(self) -> None:
 self.counter = itertools.count(1)
 self.missing_id_metadata = None

 def _get_field_id(self, field: pa.Field) -> int:
 field_id: Optional[int] = None

 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field.metadata and (field_id_str := field.metadata.get(pyarrow_field_id_key)):
 field_id = int(field_id_str.decode())

 if field_id is None:
 if self.missing_id_metadata is None:
 warnings.warn(
 "Missing field-IDs will be auto-assigned, possibly leading to inconsistencies between the file schema and the schema stored in table metadata."
 )
 field_id = next(self.counter)
 missing_is_metadata = True
 else:
 missing_is_metadata = False

def _get_field_doc(field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if doc_str := field.metadata.get(pyarrow_doc_key):
 return doc_str.decode()
 return None
 if self.missing_id_metadata is not None and self.missing_id_metadata != missing_is_metadata:
 raise ValueError("Parquet file contains partial field-ids")
 else:
 self.missing_id_metadata = missing_is_metadata

 return field_id

 def _get_field_doc(self, field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if field.metadata and (doc_str := field.metadata.get(pyarrow_doc_key)):
 return doc_str.decode()
 return None

class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 def _convert_fields(self, arrow_fields: Iterable[pa.Field], field_results: List[Optional[IcebergType]]) -> List[NestedField]:
 fields = []
 for i, field in enumerate(arrow_fields):
 field_id = _get_field_id(field)
 field_doc = _get_field_doc(field)
 field_id = self._get_field_id(field)
 field_doc = self._get_field_doc(field)
 field_type = field_results[i]
 if field_type is not None and field_id is not None:
 if field_type is not None:
 fields.append(NestedField(field_id, field.name, field_type, required=not field.nullable, doc=field_doc))
 return fields

Expand All@@ -746,7 +777,7 @@ def struct(self, struct: pa.StructType, field_results: List[Optional[IcebergType

 def list(self, list_type: pa.ListType, element_result: Optional[IcebergType]) -> Optional[IcebergType]:
 element_field = list_type.value_field
 element_id = _get_field_id(element_field)
 element_id = self._get_field_id(element_field)
 if element_result is not None and element_id is not None:
 return ListType(element_id, element_result, element_required=not element_field.nullable)
 return None
Expand All@@ -755,9 +786,9 @@ def map(
 self, map_type: pa.MapType, key_result: Optional[IcebergType], value_result: Optional[IcebergType]
 ) -> Optional[IcebergType]:
 key_field = map_type.key_field
 key_id = _get_field_id(key_field)
 key_id = self._get_field_id(key_field)
 value_field = map_type.item_field
 value_id = _get_field_id(value_field)
 value_id = self._get_field_id(value_field)
 if key_result is not None and value_result is not None and key_id is not None and value_id is not None:
 return MapType(key_id, key_result, value_id, value_result, value_required=not value_field.nullable)
 return None
Expand Down

30 changes: 30 additions & 0 deletions
tests/io/test_pyarrow_visitor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
# pylint: disable=protected-access,unused-argument,redefined-outer-name
import re
from unittest.mock import Mock, patch

import pyarrow as pa
import pytest
Expand DownExpand Up@@ -269,3 +270,32 @@ def test_round_schema_conversion_nested(table_schema_nested: Schema) -> None:
 15: person: optional struct<16: name: optional string, 17: age: required int>
}"""
 assert actual == expected


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_ids(warn: Mock) -> None:
 schema = pa.schema([pa.field('some_int', pa.int32(), nullable=True), pa.field('some_string', pa.string(), nullable=False)])
 actual = pyarrow_to_schema(schema)

 expected = Schema(
 NestedField(field_id=1, name="some_int", field_type=IntegerType(), required=False),
 NestedField(field_id=2, name="some_string", field_type=StringType(), required=True),
 )

 assert actual == expected
 assert warn.called


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_id(warn: Mock) -> None:
 schema = pa.schema(
 [
 pa.field('some_int', pa.int32(), nullable=True),
 pa.field('some_string', pa.string(), nullable=False, metadata={b"field_id": "22"}),
 ]
 )

 with pytest.raises(ValueError) as exc_info:
 _ = pyarrow_to_schema(schema)
 assert "Parquet file contains partial field-ids" in str(exc_info.value)
 assert warn.called

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default");
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Arrow: Allow missing field-ids from Schema#183

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

 Closed

Fokko
wants to merge
4
commits into

from

 Closed


Arrow: Allow missing field-ids from Schema
#183
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
405d36c

Arrow: Allow missing field-ids from Schema


Fokko Dec 5, 2023
27017cf

Thanks Honah!


Fokko Dec 7, 2023
2657fe1

Update pyiceberg/io/pyarrow.py


Fokko Dec 9, 2023
447d22f

lint


Fokko Dec 9, 2023

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


67 changes: 49 additions & 18 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,9 +25,11 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
import warnings
from abc import ABC, abstractmethod
from concurrent.futures import Future
from dataclasses import dataclass
Expand DownExpand Up@@ -110,6 +112,7 @@
 Schema,
 SchemaVisitorPerPrimitiveType,
 SchemaWithPartnerVisitor,
 assign_fresh_schema_ids,
 pre_order_visit,
 promote,
 prune_columns,
Expand DownExpand Up@@ -616,7 +619,12 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:

def pyarrow_to_schema(schema: pa.Schema) -> Schema:
 visitor = _ConvertToIceberg()
 return visit_pyarrow(schema, visitor)
 schema = visit_pyarrow(schema, visitor)

 if visitor.missing_id_metadata:
 return assign_fresh_schema_ids(schema)
 else:
 return schema


@singledispatch
Expand DownExpand Up@@ -713,28 +721,51 @@ def primitive(self, primitive: pa.DataType) -> Optional[T]:
 """Visit a primitive type."""


def _get_field_id(field: pa.Field) -> Optional[int]:
 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field_id_str := field.metadata.get(pyarrow_field_id_key):
 return int(field_id_str.decode())
 return None
class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 counter: itertools.count[int]
 missing_id_metadata: Optional[bool]

 def __init__(self) -> None:
 self.counter = itertools.count(1)
 self.missing_id_metadata = None

 def _get_field_id(self, field: pa.Field) -> int:
 field_id: Optional[int] = None

 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field.metadata and (field_id_str := field.metadata.get(pyarrow_field_id_key)):
 field_id = int(field_id_str.decode())

 if field_id is None:
 if self.missing_id_metadata is None:
 warnings.warn(
 "Missing field-IDs will be auto-assigned, possibly leading to inconsistencies between the file schema and the schema stored in table metadata."
 )
 field_id = next(self.counter)
 missing_is_metadata = True
 else:
 missing_is_metadata = False

def _get_field_doc(field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if doc_str := field.metadata.get(pyarrow_doc_key):
 return doc_str.decode()
 return None
 if self.missing_id_metadata is not None and self.missing_id_metadata != missing_is_metadata:
 raise ValueError("Parquet file contains partial field-ids")
 else:
 self.missing_id_metadata = missing_is_metadata

 return field_id

 def _get_field_doc(self, field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if field.metadata and (doc_str := field.metadata.get(pyarrow_doc_key)):
 return doc_str.decode()
 return None

class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 def _convert_fields(self, arrow_fields: Iterable[pa.Field], field_results: List[Optional[IcebergType]]) -> List[NestedField]:
 fields = []
 for i, field in enumerate(arrow_fields):
 field_id = _get_field_id(field)
 field_doc = _get_field_doc(field)
 field_id = self._get_field_id(field)
 field_doc = self._get_field_doc(field)
 field_type = field_results[i]
 if field_type is not None and field_id is not None:
 if field_type is not None:
 fields.append(NestedField(field_id, field.name, field_type, required=not field.nullable, doc=field_doc))
 return fields

Expand All@@ -746,7 +777,7 @@ def struct(self, struct: pa.StructType, field_results: List[Optional[IcebergType

 def list(self, list_type: pa.ListType, element_result: Optional[IcebergType]) -> Optional[IcebergType]:
 element_field = list_type.value_field
 element_id = _get_field_id(element_field)
 element_id = self._get_field_id(element_field)
 if element_result is not None and element_id is not None:
 return ListType(element_id, element_result, element_required=not element_field.nullable)
 return None
Expand All@@ -755,9 +786,9 @@ def map(
 self, map_type: pa.MapType, key_result: Optional[IcebergType], value_result: Optional[IcebergType]
 ) -> Optional[IcebergType]:
 key_field = map_type.key_field
 key_id = _get_field_id(key_field)
 key_id = self._get_field_id(key_field)
 value_field = map_type.item_field
 value_id = _get_field_id(value_field)
 value_id = self._get_field_id(value_field)
 if key_result is not None and value_result is not None and key_id is not None and value_id is not None:
 return MapType(key_id, key_result, value_id, value_result, value_required=not value_field.nullable)
 return None
Expand Down

30 changes: 30 additions & 0 deletions
tests/io/test_pyarrow_visitor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
# pylint: disable=protected-access,unused-argument,redefined-outer-name
import re
from unittest.mock import Mock, patch

import pyarrow as pa
import pytest
Expand DownExpand Up@@ -269,3 +270,32 @@ def test_round_schema_conversion_nested(table_schema_nested: Schema) -> None:
 15: person: optional struct<16: name: optional string, 17: age: required int>
}"""
 assert actual == expected


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_ids(warn: Mock) -> None:
 schema = pa.schema([pa.field('some_int', pa.int32(), nullable=True), pa.field('some_string', pa.string(), nullable=False)])
 actual = pyarrow_to_schema(schema)

 expected = Schema(
 NestedField(field_id=1, name="some_int", field_type=IntegerType(), required=False),
 NestedField(field_id=2, name="some_string", field_type=StringType(), required=True),
 )

 assert actual == expected
 assert warn.called


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_id(warn: Mock) -> None:
 schema = pa.schema(
 [
 pa.field('some_int', pa.int32(), nullable=True),
 pa.field('some_string', pa.string(), nullable=False, metadata={b"field_id": "22"}),
 ]
 )

 with pytest.raises(ValueError) as exc_info:
 _ = pyarrow_to_schema(schema)
 assert "Parquet file contains partial field-ids" in str(exc_info.value)
 assert warn.called

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers");
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Arrow: Allow missing field-ids from Schema#183

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

 Closed

Fokko
wants to merge
4
commits into

from

 Closed


Arrow: Allow missing field-ids from Schema
#183
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
405d36c

Arrow: Allow missing field-ids from Schema


Fokko Dec 5, 2023
27017cf

Thanks Honah!


Fokko Dec 7, 2023
2657fe1

Update pyiceberg/io/pyarrow.py


Fokko Dec 9, 2023
447d22f

lint


Fokko Dec 9, 2023

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


67 changes: 49 additions & 18 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,9 +25,11 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
import warnings
from abc import ABC, abstractmethod
from concurrent.futures import Future
from dataclasses import dataclass
Expand DownExpand Up@@ -110,6 +112,7 @@
 Schema,
 SchemaVisitorPerPrimitiveType,
 SchemaWithPartnerVisitor,
 assign_fresh_schema_ids,
 pre_order_visit,
 promote,
 prune_columns,
Expand DownExpand Up@@ -616,7 +619,12 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:

def pyarrow_to_schema(schema: pa.Schema) -> Schema:
 visitor = _ConvertToIceberg()
 return visit_pyarrow(schema, visitor)
 schema = visit_pyarrow(schema, visitor)

 if visitor.missing_id_metadata:
 return assign_fresh_schema_ids(schema)
 else:
 return schema


@singledispatch
Expand DownExpand Up@@ -713,28 +721,51 @@ def primitive(self, primitive: pa.DataType) -> Optional[T]:
 """Visit a primitive type."""


def _get_field_id(field: pa.Field) -> Optional[int]:
 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field_id_str := field.metadata.get(pyarrow_field_id_key):
 return int(field_id_str.decode())
 return None
class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 counter: itertools.count[int]
 missing_id_metadata: Optional[bool]

 def __init__(self) -> None:
 self.counter = itertools.count(1)
 self.missing_id_metadata = None

 def _get_field_id(self, field: pa.Field) -> int:
 field_id: Optional[int] = None

 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field.metadata and (field_id_str := field.metadata.get(pyarrow_field_id_key)):
 field_id = int(field_id_str.decode())

 if field_id is None:
 if self.missing_id_metadata is None:
 warnings.warn(
 "Missing field-IDs will be auto-assigned, possibly leading to inconsistencies between the file schema and the schema stored in table metadata."
 )
 field_id = next(self.counter)
 missing_is_metadata = True
 else:
 missing_is_metadata = False

def _get_field_doc(field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if doc_str := field.metadata.get(pyarrow_doc_key):
 return doc_str.decode()
 return None
 if self.missing_id_metadata is not None and self.missing_id_metadata != missing_is_metadata:
 raise ValueError("Parquet file contains partial field-ids")
 else:
 self.missing_id_metadata = missing_is_metadata

 return field_id

 def _get_field_doc(self, field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if field.metadata and (doc_str := field.metadata.get(pyarrow_doc_key)):
 return doc_str.decode()
 return None

class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 def _convert_fields(self, arrow_fields: Iterable[pa.Field], field_results: List[Optional[IcebergType]]) -> List[NestedField]:
 fields = []
 for i, field in enumerate(arrow_fields):
 field_id = _get_field_id(field)
 field_doc = _get_field_doc(field)
 field_id = self._get_field_id(field)
 field_doc = self._get_field_doc(field)
 field_type = field_results[i]
 if field_type is not None and field_id is not None:
 if field_type is not None:
 fields.append(NestedField(field_id, field.name, field_type, required=not field.nullable, doc=field_doc))
 return fields

Expand All@@ -746,7 +777,7 @@ def struct(self, struct: pa.StructType, field_results: List[Optional[IcebergType

 def list(self, list_type: pa.ListType, element_result: Optional[IcebergType]) -> Optional[IcebergType]:
 element_field = list_type.value_field
 element_id = _get_field_id(element_field)
 element_id = self._get_field_id(element_field)
 if element_result is not None and element_id is not None:
 return ListType(element_id, element_result, element_required=not element_field.nullable)
 return None
Expand All@@ -755,9 +786,9 @@ def map(
 self, map_type: pa.MapType, key_result: Optional[IcebergType], value_result: Optional[IcebergType]
 ) -> Optional[IcebergType]:
 key_field = map_type.key_field
 key_id = _get_field_id(key_field)
 key_id = self._get_field_id(key_field)
 value_field = map_type.item_field
 value_id = _get_field_id(value_field)
 value_id = self._get_field_id(value_field)
 if key_result is not None and value_result is not None and key_id is not None and value_id is not None:
 return MapType(key_id, key_result, value_id, value_result, value_required=not value_field.nullable)
 return None
Expand Down

30 changes: 30 additions & 0 deletions
tests/io/test_pyarrow_visitor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
# pylint: disable=protected-access,unused-argument,redefined-outer-name
import re
from unittest.mock import Mock, patch

import pyarrow as pa
import pytest
Expand DownExpand Up@@ -269,3 +270,32 @@ def test_round_schema_conversion_nested(table_schema_nested: Schema) -> None:
 15: person: optional struct<16: name: optional string, 17: age: required int>
}"""
 assert actual == expected


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_ids(warn: Mock) -> None:
 schema = pa.schema([pa.field('some_int', pa.int32(), nullable=True), pa.field('some_string', pa.string(), nullable=False)])
 actual = pyarrow_to_schema(schema)

 expected = Schema(
 NestedField(field_id=1, name="some_int", field_type=IntegerType(), required=False),
 NestedField(field_id=2, name="some_string", field_type=StringType(), required=True),
 )

 assert actual == expected
 assert warn.called


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_id(warn: Mock) -> None:
 schema = pa.schema(
 [
 pa.field('some_int', pa.int32(), nullable=True),
 pa.field('some_string', pa.string(), nullable=False, metadata={b"field_id": "22"}),
 ]
 )

 with pytest.raises(ValueError) as exc_info:
 _ = pyarrow_to_schema(schema)
 assert "Parquet file contains partial field-ids" in str(exc_info.value)
 assert warn.called

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode");
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
Arrow: Allow missing field-ids from Schema#183

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

 Closed

Fokko
wants to merge
4
commits into

from

 Closed


Arrow: Allow missing field-ids from Schema
#183
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


4 commits


Select commit
Hold shift + click to select a range
405d36c

Arrow: Allow missing field-ids from Schema


Fokko Dec 5, 2023
27017cf

Thanks Honah!


Fokko Dec 7, 2023
2657fe1

Update pyiceberg/io/pyarrow.py


Fokko Dec 9, 2023
447d22f

lint


Fokko Dec 9, 2023

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


67 changes: 49 additions & 18 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -25,9 +25,11 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
import warnings
from abc import ABC, abstractmethod
from concurrent.futures import Future
from dataclasses import dataclass
Expand DownExpand Up@@ -110,6 +112,7 @@
 Schema,
 SchemaVisitorPerPrimitiveType,
 SchemaWithPartnerVisitor,
 assign_fresh_schema_ids,
 pre_order_visit,
 promote,
 prune_columns,
Expand DownExpand Up@@ -616,7 +619,12 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:

def pyarrow_to_schema(schema: pa.Schema) -> Schema:
 visitor = _ConvertToIceberg()
 return visit_pyarrow(schema, visitor)
 schema = visit_pyarrow(schema, visitor)

 if visitor.missing_id_metadata:
 return assign_fresh_schema_ids(schema)
 else:
 return schema


@singledispatch
Expand DownExpand Up@@ -713,28 +721,51 @@ def primitive(self, primitive: pa.DataType) -> Optional[T]:
 """Visit a primitive type."""


def _get_field_id(field: pa.Field) -> Optional[int]:
 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field_id_str := field.metadata.get(pyarrow_field_id_key):
 return int(field_id_str.decode())
 return None
class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 counter: itertools.count[int]
 missing_id_metadata: Optional[bool]

 def __init__(self) -> None:
 self.counter = itertools.count(1)
 self.missing_id_metadata = None

 def _get_field_id(self, field: pa.Field) -> int:
 field_id: Optional[int] = None

 for pyarrow_field_id_key in PYARROW_FIELD_ID_KEYS:
 if field.metadata and (field_id_str := field.metadata.get(pyarrow_field_id_key)):
 field_id = int(field_id_str.decode())

 if field_id is None:
 if self.missing_id_metadata is None:
 warnings.warn(
 "Missing field-IDs will be auto-assigned, possibly leading to inconsistencies between the file schema and the schema stored in table metadata."
 )
 field_id = next(self.counter)
 missing_is_metadata = True
 else:
 missing_is_metadata = False

def _get_field_doc(field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if doc_str := field.metadata.get(pyarrow_doc_key):
 return doc_str.decode()
 return None
 if self.missing_id_metadata is not None and self.missing_id_metadata != missing_is_metadata:
 raise ValueError("Parquet file contains partial field-ids")
 else:
 self.missing_id_metadata = missing_is_metadata

 return field_id

 def _get_field_doc(self, field: pa.Field) -> Optional[str]:
 for pyarrow_doc_key in PYARROW_FIELD_DOC_KEYS:
 if field.metadata and (doc_str := field.metadata.get(pyarrow_doc_key)):
 return doc_str.decode()
 return None

class _ConvertToIceberg(PyArrowSchemaVisitor[Union[IcebergType, Schema]]):
 def _convert_fields(self, arrow_fields: Iterable[pa.Field], field_results: List[Optional[IcebergType]]) -> List[NestedField]:
 fields = []
 for i, field in enumerate(arrow_fields):
 field_id = _get_field_id(field)
 field_doc = _get_field_doc(field)
 field_id = self._get_field_id(field)
 field_doc = self._get_field_doc(field)
 field_type = field_results[i]
 if field_type is not None and field_id is not None:
 if field_type is not None:
 fields.append(NestedField(field_id, field.name, field_type, required=not field.nullable, doc=field_doc))
 return fields

Expand All@@ -746,7 +777,7 @@ def struct(self, struct: pa.StructType, field_results: List[Optional[IcebergType

 def list(self, list_type: pa.ListType, element_result: Optional[IcebergType]) -> Optional[IcebergType]:
 element_field = list_type.value_field
 element_id = _get_field_id(element_field)
 element_id = self._get_field_id(element_field)
 if element_result is not None and element_id is not None:
 return ListType(element_id, element_result, element_required=not element_field.nullable)
 return None
Expand All@@ -755,9 +786,9 @@ def map(
 self, map_type: pa.MapType, key_result: Optional[IcebergType], value_result: Optional[IcebergType]
 ) -> Optional[IcebergType]:
 key_field = map_type.key_field
 key_id = _get_field_id(key_field)
 key_id = self._get_field_id(key_field)
 value_field = map_type.item_field
 value_id = _get_field_id(value_field)
 value_id = self._get_field_id(value_field)
 if key_result is not None and value_result is not None and key_id is not None and value_id is not None:
 return MapType(key_id, key_result, value_id, value_result, value_required=not value_field.nullable)
 return None
Expand Down

30 changes: 30 additions & 0 deletions
tests/io/test_pyarrow_visitor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
# pylint: disable=protected-access,unused-argument,redefined-outer-name
import re
from unittest.mock import Mock, patch

import pyarrow as pa
import pytest
Expand DownExpand Up@@ -269,3 +270,32 @@ def test_round_schema_conversion_nested(table_schema_nested: Schema) -> None:
 15: person: optional struct<16: name: optional string, 17: age: required int>
}"""
 assert actual == expected


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_ids(warn: Mock) -> None:
 schema = pa.schema([pa.field('some_int', pa.int32(), nullable=True), pa.field('some_string', pa.string(), nullable=False)])
 actual = pyarrow_to_schema(schema)

 expected = Schema(
 NestedField(field_id=1, name="some_int", field_type=IntegerType(), required=False),
 NestedField(field_id=2, name="some_string", field_type=StringType(), required=True),
 )

 assert actual == expected
 assert warn.called


@patch("warnings.warn")
def test_schema_to_pyarrow_schema_missing_id(warn: Mock) -> None:
 schema = pa.schema(
 [
 pa.field('some_int', pa.int32(), nullable=True),
 pa.field('some_string', pa.string(), nullable=False, metadata={b"field_id": "22"}),
 ]
 )

 with pytest.raises(ValueError) as exc_info:
 _ = pyarrow_to_schema(schema)
 assert "Parquet file contains partial field-ids" in str(exc_info.value)
 assert warn.called