create_table with a PyArrow Schema#305

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
HonahX merged 9 commits into from
Jan 30, 2024
Merged

create_table with a PyArrow Schema #305

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
8cd0e52
create_table from pyarrow schema
sungwy Jan 25, 2024
3d0445b
incorporate review comments
sungwy Jan 25, 2024
6ea0892
docs
sungwy Jan 25, 2024
7ba0fd1
Take suggestion
sungwy Jan 27, 2024
39a97a2
Take suggestion
sungwy Jan 27, 2024
60ac8f8
take nit
sungwy Jan 27, 2024
b917afd
more nit
sungwy Jan 27, 2024
e4e9f9b
refactoring
sungwy Jan 29, 2024
c40c553
adopt review suggestions
sungwy Jan 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
19 changes: 19 additions & 0 deletions mkdocs/docs/api.md
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -146,6 +146,25 @@ catalog.create_table(
)
```

To create a table using a pyarrow schema:

```python
import pyarrow as pa

schema = pa.schema(
[
pa.field("foo", pa.string(), nullable=True),
pa.field("bar", pa.int32(), nullable=False),
pa.field("baz", pa.bool_(), nullable=True),
]
)

catalog.create_table(
identifier="docs_example.bids",
schema=schema,
)
```

## Load a table

### Catalog table
Expand Down
22 changes: 21 additions & 1 deletion pyiceberg/catalog/__init__.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,6 +24,7 @@
from dataclasses import dataclass
from enum import Enum
from typing import (
TYPE_CHECKING,
Callable,
Dict,
List,
Expand DownExpand Up@@ -56,6 +57,9 @@
)
from pyiceberg.utils.config import Config, merge_config

if TYPE_CHECKING:
import pyarrow as pa

logger = logging.getLogger(__name__)

_ENV_CONFIG = Config()
Expand DownExpand Up@@ -288,7 +292,7 @@ def _load_file_io(self, properties: Properties = EMPTY_DICT, location: Optional[
def create_table(
self,
identifier: Union[str, Identifier],
schema: Schema,
schema: Union[Schema, "pa.Schema"],
location: Optional[str] = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand DownExpand Up@@ -512,6 +516,22 @@ def _check_for_overlap(removals: Optional[Set[str]], updates: Properties) -> Non
if overlap:
raise ValueError(f"Updates and deletes have an overlap: {overlap}")

@staticmethod
def _convert_schema_if_needed(schema: Union[Schema, "pa.Schema"]) -> Schema:
if isinstance(schema, Schema):
return schema
try:
Comment thread
HonahX marked this conversation as resolved.
import pyarrow as pa

from pyiceberg.io.pyarrow import _ConvertToIcebergWithoutIDs, visit_pyarrow

if isinstance(schema, pa.Schema):
schema: Schema = visit_pyarrow(schema, _ConvertToIcebergWithoutIDs()) # type: ignore
return schema
except ModuleNotFoundError:
pass
raise ValueError(f"{type(schema)=}, but it must be pyiceberg.schema.Schema or pyarrow.Schema")

def _resolve_table_location(self, location: Optional[str], database_name: str, table_name: str) -> str:
if not location:
return self._get_default_warehouse_location(database_name, table_name)
Expand Down
8 changes: 7 additions & 1 deletion pyiceberg/catalog/dynamodb.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@
import uuid
from time import time
from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
Expand DownExpand Up@@ -57,6 +58,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
import pyarrow as pa

DYNAMODB_CLIENT = "dynamodb"

DYNAMODB_COL_IDENTIFIER = "identifier"
Expand DownExpand Up@@ -127,7 +131,7 @@ def _dynamodb_table_exists(self) -> bool:
def create_table(
self,
identifier: Union[str, Identifier],
schema: Schema,
schema: Union[Schema, "pa.Schema"],
location: Optional[str] = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -152,6 +156,8 @@ def create_table(
ValueError: If the identifier is invalid, or no path is given to store metadata.

"""
schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

database_name, table_name = self.identifier_to_database_and_table(identifier)

location = self._resolve_table_location(location, database_name, table_name)
Expand Down
8 changes: 7 additions & 1 deletion pyiceberg/catalog/glue.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@


from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
Expand DownExpand Up@@ -88,6 +89,9 @@
UUIDType,
)

if TYPE_CHECKING:
import pyarrow as pa

# If Glue should skip archiving an old table version when creating a new version in a commit. By
# default, Glue archives all old table versions after an UpdateTable call, but Glue has a default
# max number of archived table versions (can be increased). So for streaming use case with lots
Expand DownExpand Up@@ -329,7 +333,7 @@ def _get_glue_table(self, database_name: str, table_name: str) -> TableTypeDef:
def create_table(
self,
identifier: Union[str, Identifier],
schema: Schema,
schema: Union[Schema, "pa.Schema"],
location: Optional[str] = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -354,6 +358,8 @@ def create_table(
ValueError: If the identifier is invalid, or no path is given to store metadata.

"""
schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

database_name, table_name = self.identifier_to_database_and_table(identifier)

location = self._resolve_table_location(location, database_name, table_name)
Expand Down
9 changes: 8 additions & 1 deletion pyiceberg/catalog/hive.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -18,6 +18,7 @@
import time
from types import TracebackType
from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
Expand DownExpand Up@@ -91,6 +92,10 @@
UUIDType,
)

if TYPE_CHECKING:
import pyarrow as pa


# Replace by visitor
hive_types = {
BooleanType: "boolean",
Expand DownExpand Up@@ -250,7 +255,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable, io: FileIO) -> Table:
def create_table(
self,
identifier: Union[str, Identifier],
schema: Schema,
schema: Union[Schema, "pa.Schema"],
location: Optional[str] = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -273,6 +278,8 @@ def create_table(
AlreadyExistsError: If a table with the name already exists.
ValueError: If the identifier is invalid.
"""
schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

properties = {**DEFAULT_PROPERTIES, **properties}
database_name, table_name = self.identifier_to_database_and_table(identifier)
current_time_millis = int(time.time() * 1000)
Expand Down
6 changes: 5 additions & 1 deletion pyiceberg/catalog/noop.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
from typing import (
TYPE_CHECKING,
List,
Optional,
Set,
Expand All@@ -33,12 +34,15 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties

if TYPE_CHECKING:
import pyarrow as pa


class NoopCatalog(Catalog):
def create_table(
self,
identifier: Union[str, Identifier],
schema: Schema,
schema: Union[Schema, "pa.Schema"],
location: Optional[str] = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand Down
8 changes: 7 additions & 1 deletion pyiceberg/catalog/rest.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
from json import JSONDecodeError
from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
Expand DownExpand Up@@ -68,6 +69,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel

if TYPE_CHECKING:
import pyarrow as pa

ICEBERG_REST_SPEC_VERSION = "0.14.1"


Expand DownExpand Up@@ -437,12 +441,14 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
def create_table(
self,
identifier: Union[str, Identifier],
schema: Schema,
schema: Union[Schema, "pa.Schema"],
location: Optional[str] = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
properties: Properties = EMPTY_DICT,
) -> Table:
schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

namespace_and_table = self._split_identifier_for_path(identifier)
request = CreateTableRequest(
name=namespace_and_table["table"],
Expand Down
8 changes: 7 additions & 1 deletion pyiceberg/catalog/sql.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.

from typing import (
TYPE_CHECKING,
List,
Optional,
Set,
Expand DownExpand Up@@ -65,6 +66,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
import pyarrow as pa


class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase):
pass
Expand DownExpand Up@@ -140,7 +144,7 @@ def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table:
def create_table(
self,
identifier: Union[str, Identifier],
schema: Schema,
schema: Union[Schema, "pa.Schema"],
location: Optional[str] = None,
partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -165,6 +169,8 @@ def create_table(
ValueError: If the identifier is invalid, or no path is given to store metadata.

"""
schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

database_name, table_name = self.identifier_to_database_and_table(identifier)
if not self._namespace_exists(database_name):
raise NoSuchNamespaceError(f"Namespace does not exist: {database_name}")
Expand Down
25 changes: 20 additions & 5 deletions pyiceberg/io/pyarrow.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,7 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
Expand All@@ -34,7 +35,6 @@
from dataclasses import dataclass
from enum import Enum
from functools import lru_cache, singledispatch
from itertools import chain
from typing import (
TYPE_CHECKING,
Any,
Expand DownExpand Up@@ -631,7 +631,7 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:
if len(positional_deletes) == 1:
all_chunks = positional_deletes[0]
else:
all_chunks = pa.chunked_array(chain(*[arr.chunks for arr in positional_deletes]))
all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
return np.setdiff1d(np.arange(rows), all_chunks, assume_unique=False)


Expand DownExpand Up@@ -906,6 +906,21 @@ def after_map_value(self, element: pa.Field) -> None:
self._field_names.pop()


class _ConvertToIcebergWithoutIDs(_ConvertToIceberg):
"""
Converts PyArrowSchema to Iceberg Schema with all -1 ids.

The schema generated through this visitor should always be
used in conjunction with `new_table_metadata` function to
assign new field ids in order. This is currently used only
when creating an Iceberg Schema from a PyArrow schema when
creating a new Iceberg table.
"""

def _field_id(self, field: pa.Field) -> int:
return -1


def _task_to_table(
fs: FileSystem,
task: FileScanTask,
Expand DownExpand Up@@ -993,7 +1008,7 @@ def _task_to_table(

def _read_all_delete_files(fs: FileSystem, tasks: Iterable[FileScanTask]) -> Dict[str, List[ChunkedArray]]:
deletes_per_file: Dict[str, List[ChunkedArray]] = {}
unique_deletes = set(chain.from_iterable([task.delete_files for task in tasks]))
unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks]))
if len(unique_deletes) > 0:
executor = ExecutorFactory.get_or_create()
deletes_per_files: Iterator[Dict[str, ChunkedArray]] = executor.map(
Expand DownExpand Up@@ -1399,7 +1414,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[StatisticsColl
def struct(
self, struct: StructType, field_results: List[Callable[[], List[StatisticsCollector]]]
) -> List[StatisticsCollector]:
return list(chain(*[result() for result in field_results]))
return list(itertools.chain(*[result() for result in field_results]))

def field(self, field: NestedField, field_result: Callable[[], List[StatisticsCollector]]) -> List[StatisticsCollector]:
self._field_id = field.field_id
Expand DownExpand Up@@ -1491,7 +1506,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[ID2ParquetPath
return struct_result()

def struct(self, struct: StructType, field_results: List[Callable[[], List[ID2ParquetPath]]]) -> List[ID2ParquetPath]:
return list(chain(*[result() for result in field_results]))
return list(itertools.chain(*[result() for result in field_results]))

def field(self, field: NestedField, field_result: Callable[[], List[ID2ParquetPath]]) -> List[ID2ParquetPath]:
self._field_id = field.field_id
Expand Down
Loading
, '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
create_table with a PyArrow Schema#305

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

HonahX
merged 9 commits into

from

Jan 30, 2024
 Merged

create_table with a PyArrow Schema
#305
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


9 commits


Select commit
Hold shift + click to select a range
8cd0e52

create_table from pyarrow schema


sungwy Jan 25, 2024
3d0445b

incorporate review comments


sungwy Jan 25, 2024
6ea0892

docs


sungwy Jan 25, 2024
7ba0fd1

Take suggestion


sungwy Jan 27, 2024
39a97a2

Take suggestion


sungwy Jan 27, 2024
60ac8f8

take nit


sungwy Jan 27, 2024
b917afd

more nit


sungwy Jan 27, 2024
e4e9f9b

refactoring


sungwy Jan 29, 2024
c40c553

adopt review suggestions


sungwy Jan 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


19 changes: 19 additions & 0 deletions
mkdocs/docs/api.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -146,6 +146,25 @@ catalog.create_table(
)
```

To create a table using a pyarrow schema:

```python
import pyarrow as pa

schema = pa.schema(
 [
 pa.field("foo", pa.string(), nullable=True),
 pa.field("bar", pa.int32(), nullable=False),
 pa.field("baz", pa.bool_(), nullable=True),
 ]
)

catalog.create_table(
 identifier="docs_example.bids",
 schema=schema,
)
```

## Load a table

### Catalog table
Expand Down

22 changes: 21 additions & 1 deletion
pyiceberg/catalog/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,6 +24,7 @@
from dataclasses import dataclass
from enum import Enum
from typing import (
 TYPE_CHECKING,
 Callable,
 Dict,
 List,
Expand DownExpand Up@@ -56,6 +57,9 @@
)
from pyiceberg.utils.config import Config, merge_config

if TYPE_CHECKING:
 import pyarrow as pa

logger = logging.getLogger(__name__)

_ENV_CONFIG = Config()
Expand DownExpand Up@@ -288,7 +292,7 @@ def _load_file_io(self, properties: Properties = EMPTY_DICT, location: Optional[
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand DownExpand Up@@ -512,6 +516,22 @@ def _check_for_overlap(removals: Optional[Set[str]], updates: Properties) -> Non
 if overlap:
 raise ValueError(f"Updates and deletes have an overlap: {overlap}")

 @staticmethod
 def _convert_schema_if_needed(schema: Union[Schema, "pa.Schema"]) -> Schema:
 if isinstance(schema, Schema):
 return schema
 try:
Comment thread
HonahX marked this conversation as resolved.

 import pyarrow as pa

 from pyiceberg.io.pyarrow import _ConvertToIcebergWithoutIDs, visit_pyarrow

 if isinstance(schema, pa.Schema):
 schema: Schema = visit_pyarrow(schema, _ConvertToIcebergWithoutIDs()) # type: ignore
 return schema
 except ModuleNotFoundError:
 pass
 raise ValueError(f"{type(schema)=}, but it must be pyiceberg.schema.Schema or pyarrow.Schema")

 def _resolve_table_location(self, location: Optional[str], database_name: str, table_name: str) -> str:
 if not location:
 return self._get_default_warehouse_location(database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/dynamodb.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@
import uuid
from time import time
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -57,6 +58,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa

DYNAMODB_CLIENT = "dynamodb"

DYNAMODB_COL_IDENTIFIER = "identifier"
Expand DownExpand Up@@ -127,7 +131,7 @@ def _dynamodb_table_exists(self) -> bool:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -152,6 +156,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/glue.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@


from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -88,6 +89,9 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa

# If Glue should skip archiving an old table version when creating a new version in a commit. By
# default, Glue archives all old table versions after an UpdateTable call, but Glue has a default
# max number of archived table versions (can be increased). So for streaming use case with lots
Expand DownExpand Up@@ -329,7 +333,7 @@ def _get_glue_table(self, database_name: str, table_name: str) -> TableTypeDef:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -354,6 +358,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

9 changes: 8 additions & 1 deletion
pyiceberg/catalog/hive.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -18,6 +18,7 @@
import time
from types import TracebackType
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -91,6 +92,10 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa


# Replace by visitor
hive_types = {
 BooleanType: "boolean",
Expand DownExpand Up@@ -250,7 +255,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable, io: FileIO) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -273,6 +278,8 @@ def create_table(
 AlreadyExistsError: If a table with the name already exists.
 ValueError: If the identifier is invalid.
 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 properties = {**DEFAULT_PROPERTIES, **properties}
 database_name, table_name = self.identifier_to_database_and_table(identifier)
 current_time_millis = int(time.time() * 1000)
Expand Down

6 changes: 5 additions & 1 deletion
pyiceberg/catalog/noop.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand All@@ -33,12 +34,15 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties

if TYPE_CHECKING:
 import pyarrow as pa


class NoopCatalog(Catalog):
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/rest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
from json import JSONDecodeError
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -68,6 +69,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel

if TYPE_CHECKING:
 import pyarrow as pa

ICEBERG_REST_SPEC_VERSION = "0.14.1"


Expand DownExpand Up@@ -437,12 +441,14 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
 properties: Properties = EMPTY_DICT,
 ) -> Table:
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 namespace_and_table = self._split_identifier_for_path(identifier)
 request = CreateTableRequest(
 name=namespace_and_table["table"],
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.

from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand DownExpand Up@@ -65,6 +66,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa


class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase):
 pass
Expand DownExpand Up@@ -140,7 +144,7 @@ def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -165,6 +169,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)
 if not self._namespace_exists(database_name):
 raise NoSuchNamespaceError(f"Namespace does not exist: {database_name}")
Expand Down

25 changes: 20 additions & 5 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,7 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
Expand All@@ -34,7 +35,6 @@
from dataclasses import dataclass
from enum import Enum
from functools import lru_cache, singledispatch
from itertools import chain
from typing import (
 TYPE_CHECKING,
 Any,
Expand DownExpand Up@@ -631,7 +631,7 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:
 if len(positional_deletes) == 1:
 all_chunks = positional_deletes[0]
 else:
 all_chunks = pa.chunked_array(chain(*[arr.chunks for arr in positional_deletes]))
 all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
 return np.setdiff1d(np.arange(rows), all_chunks, assume_unique=False)


Expand DownExpand Up@@ -906,6 +906,21 @@ def after_map_value(self, element: pa.Field) -> None:
 self._field_names.pop()


class _ConvertToIcebergWithoutIDs(_ConvertToIceberg):
 """
 Converts PyArrowSchema to Iceberg Schema with all -1 ids.

 The schema generated through this visitor should always be
 used in conjunction with `new_table_metadata` function to
 assign new field ids in order. This is currently used only
 when creating an Iceberg Schema from a PyArrow schema when
 creating a new Iceberg table.
 """

 def _field_id(self, field: pa.Field) -> int:
 return -1


def _task_to_table(
 fs: FileSystem,
 task: FileScanTask,
Expand DownExpand Up@@ -993,7 +1008,7 @@ def _task_to_table(

def _read_all_delete_files(fs: FileSystem, tasks: Iterable[FileScanTask]) -> Dict[str, List[ChunkedArray]]:
 deletes_per_file: Dict[str, List[ChunkedArray]] = {}
 unique_deletes = set(chain.from_iterable([task.delete_files for task in tasks]))
 unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks]))
 if len(unique_deletes) > 0:
 executor = ExecutorFactory.get_or_create()
 deletes_per_files: Iterator[Dict[str, ChunkedArray]] = executor.map(
Expand DownExpand Up@@ -1399,7 +1414,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[StatisticsColl
 def struct(
 self, struct: StructType, field_results: List[Callable[[], List[StatisticsCollector]]]
 ) -> List[StatisticsCollector]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[StatisticsCollector]]) -> List[StatisticsCollector]:
 self._field_id = field.field_id
Expand DownExpand Up@@ -1491,7 +1506,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[ID2ParquetPath
 return struct_result()

 def struct(self, struct: StructType, field_results: List[Callable[[], List[ID2ParquetPath]]]) -> List[ID2ParquetPath]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[ID2ParquetPath]]) -> List[ID2ParquetPath]:
 self._field_id = field.field_id
Expand Down
Loading

, '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
create_table with a PyArrow Schema#305

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

HonahX
merged 9 commits into

from

Jan 30, 2024
 Merged

create_table with a PyArrow Schema
#305
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


9 commits


Select commit
Hold shift + click to select a range
8cd0e52

create_table from pyarrow schema


sungwy Jan 25, 2024
3d0445b

incorporate review comments


sungwy Jan 25, 2024
6ea0892

docs


sungwy Jan 25, 2024
7ba0fd1

Take suggestion


sungwy Jan 27, 2024
39a97a2

Take suggestion


sungwy Jan 27, 2024
60ac8f8

take nit


sungwy Jan 27, 2024
b917afd

more nit


sungwy Jan 27, 2024
e4e9f9b

refactoring


sungwy Jan 29, 2024
c40c553

adopt review suggestions


sungwy Jan 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


19 changes: 19 additions & 0 deletions
mkdocs/docs/api.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -146,6 +146,25 @@ catalog.create_table(
)
```

To create a table using a pyarrow schema:

```python
import pyarrow as pa

schema = pa.schema(
 [
 pa.field("foo", pa.string(), nullable=True),
 pa.field("bar", pa.int32(), nullable=False),
 pa.field("baz", pa.bool_(), nullable=True),
 ]
)

catalog.create_table(
 identifier="docs_example.bids",
 schema=schema,
)
```

## Load a table

### Catalog table
Expand Down

22 changes: 21 additions & 1 deletion
pyiceberg/catalog/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,6 +24,7 @@
from dataclasses import dataclass
from enum import Enum
from typing import (
 TYPE_CHECKING,
 Callable,
 Dict,
 List,
Expand DownExpand Up@@ -56,6 +57,9 @@
)
from pyiceberg.utils.config import Config, merge_config

if TYPE_CHECKING:
 import pyarrow as pa

logger = logging.getLogger(__name__)

_ENV_CONFIG = Config()
Expand DownExpand Up@@ -288,7 +292,7 @@ def _load_file_io(self, properties: Properties = EMPTY_DICT, location: Optional[
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand DownExpand Up@@ -512,6 +516,22 @@ def _check_for_overlap(removals: Optional[Set[str]], updates: Properties) -> Non
 if overlap:
 raise ValueError(f"Updates and deletes have an overlap: {overlap}")

 @staticmethod
 def _convert_schema_if_needed(schema: Union[Schema, "pa.Schema"]) -> Schema:
 if isinstance(schema, Schema):
 return schema
 try:
Comment thread
HonahX marked this conversation as resolved.

 import pyarrow as pa

 from pyiceberg.io.pyarrow import _ConvertToIcebergWithoutIDs, visit_pyarrow

 if isinstance(schema, pa.Schema):
 schema: Schema = visit_pyarrow(schema, _ConvertToIcebergWithoutIDs()) # type: ignore
 return schema
 except ModuleNotFoundError:
 pass
 raise ValueError(f"{type(schema)=}, but it must be pyiceberg.schema.Schema or pyarrow.Schema")

 def _resolve_table_location(self, location: Optional[str], database_name: str, table_name: str) -> str:
 if not location:
 return self._get_default_warehouse_location(database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/dynamodb.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@
import uuid
from time import time
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -57,6 +58,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa

DYNAMODB_CLIENT = "dynamodb"

DYNAMODB_COL_IDENTIFIER = "identifier"
Expand DownExpand Up@@ -127,7 +131,7 @@ def _dynamodb_table_exists(self) -> bool:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -152,6 +156,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/glue.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@


from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -88,6 +89,9 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa

# If Glue should skip archiving an old table version when creating a new version in a commit. By
# default, Glue archives all old table versions after an UpdateTable call, but Glue has a default
# max number of archived table versions (can be increased). So for streaming use case with lots
Expand DownExpand Up@@ -329,7 +333,7 @@ def _get_glue_table(self, database_name: str, table_name: str) -> TableTypeDef:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -354,6 +358,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

9 changes: 8 additions & 1 deletion
pyiceberg/catalog/hive.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -18,6 +18,7 @@
import time
from types import TracebackType
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -91,6 +92,10 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa


# Replace by visitor
hive_types = {
 BooleanType: "boolean",
Expand DownExpand Up@@ -250,7 +255,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable, io: FileIO) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -273,6 +278,8 @@ def create_table(
 AlreadyExistsError: If a table with the name already exists.
 ValueError: If the identifier is invalid.
 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 properties = {**DEFAULT_PROPERTIES, **properties}
 database_name, table_name = self.identifier_to_database_and_table(identifier)
 current_time_millis = int(time.time() * 1000)
Expand Down

6 changes: 5 additions & 1 deletion
pyiceberg/catalog/noop.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand All@@ -33,12 +34,15 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties

if TYPE_CHECKING:
 import pyarrow as pa


class NoopCatalog(Catalog):
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/rest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
from json import JSONDecodeError
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -68,6 +69,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel

if TYPE_CHECKING:
 import pyarrow as pa

ICEBERG_REST_SPEC_VERSION = "0.14.1"


Expand DownExpand Up@@ -437,12 +441,14 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
 properties: Properties = EMPTY_DICT,
 ) -> Table:
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 namespace_and_table = self._split_identifier_for_path(identifier)
 request = CreateTableRequest(
 name=namespace_and_table["table"],
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.

from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand DownExpand Up@@ -65,6 +66,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa


class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase):
 pass
Expand DownExpand Up@@ -140,7 +144,7 @@ def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -165,6 +169,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)
 if not self._namespace_exists(database_name):
 raise NoSuchNamespaceError(f"Namespace does not exist: {database_name}")
Expand Down

25 changes: 20 additions & 5 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,7 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
Expand All@@ -34,7 +35,6 @@
from dataclasses import dataclass
from enum import Enum
from functools import lru_cache, singledispatch
from itertools import chain
from typing import (
 TYPE_CHECKING,
 Any,
Expand DownExpand Up@@ -631,7 +631,7 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:
 if len(positional_deletes) == 1:
 all_chunks = positional_deletes[0]
 else:
 all_chunks = pa.chunked_array(chain(*[arr.chunks for arr in positional_deletes]))
 all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
 return np.setdiff1d(np.arange(rows), all_chunks, assume_unique=False)


Expand DownExpand Up@@ -906,6 +906,21 @@ def after_map_value(self, element: pa.Field) -> None:
 self._field_names.pop()


class _ConvertToIcebergWithoutIDs(_ConvertToIceberg):
 """
 Converts PyArrowSchema to Iceberg Schema with all -1 ids.

 The schema generated through this visitor should always be
 used in conjunction with `new_table_metadata` function to
 assign new field ids in order. This is currently used only
 when creating an Iceberg Schema from a PyArrow schema when
 creating a new Iceberg table.
 """

 def _field_id(self, field: pa.Field) -> int:
 return -1


def _task_to_table(
 fs: FileSystem,
 task: FileScanTask,
Expand DownExpand Up@@ -993,7 +1008,7 @@ def _task_to_table(

def _read_all_delete_files(fs: FileSystem, tasks: Iterable[FileScanTask]) -> Dict[str, List[ChunkedArray]]:
 deletes_per_file: Dict[str, List[ChunkedArray]] = {}
 unique_deletes = set(chain.from_iterable([task.delete_files for task in tasks]))
 unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks]))
 if len(unique_deletes) > 0:
 executor = ExecutorFactory.get_or_create()
 deletes_per_files: Iterator[Dict[str, ChunkedArray]] = executor.map(
Expand DownExpand Up@@ -1399,7 +1414,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[StatisticsColl
 def struct(
 self, struct: StructType, field_results: List[Callable[[], List[StatisticsCollector]]]
 ) -> List[StatisticsCollector]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[StatisticsCollector]]) -> List[StatisticsCollector]:
 self._field_id = field.field_id
Expand DownExpand Up@@ -1491,7 +1506,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[ID2ParquetPath
 return struct_result()

 def struct(self, struct: StructType, field_results: List[Callable[[], List[ID2ParquetPath]]]) -> List[ID2ParquetPath]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[ID2ParquetPath]]) -> List[ID2ParquetPath]:
 self._field_id = field.field_id
Expand Down
Loading

, '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
create_table with a PyArrow Schema#305

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

HonahX
merged 9 commits into

from

Jan 30, 2024
 Merged

create_table with a PyArrow Schema
#305
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


9 commits


Select commit
Hold shift + click to select a range
8cd0e52

create_table from pyarrow schema


sungwy Jan 25, 2024
3d0445b

incorporate review comments


sungwy Jan 25, 2024
6ea0892

docs


sungwy Jan 25, 2024
7ba0fd1

Take suggestion


sungwy Jan 27, 2024
39a97a2

Take suggestion


sungwy Jan 27, 2024
60ac8f8

take nit


sungwy Jan 27, 2024
b917afd

more nit


sungwy Jan 27, 2024
e4e9f9b

refactoring


sungwy Jan 29, 2024
c40c553

adopt review suggestions


sungwy Jan 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


19 changes: 19 additions & 0 deletions
mkdocs/docs/api.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -146,6 +146,25 @@ catalog.create_table(
)
```

To create a table using a pyarrow schema:

```python
import pyarrow as pa

schema = pa.schema(
 [
 pa.field("foo", pa.string(), nullable=True),
 pa.field("bar", pa.int32(), nullable=False),
 pa.field("baz", pa.bool_(), nullable=True),
 ]
)

catalog.create_table(
 identifier="docs_example.bids",
 schema=schema,
)
```

## Load a table

### Catalog table
Expand Down

22 changes: 21 additions & 1 deletion
pyiceberg/catalog/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,6 +24,7 @@
from dataclasses import dataclass
from enum import Enum
from typing import (
 TYPE_CHECKING,
 Callable,
 Dict,
 List,
Expand DownExpand Up@@ -56,6 +57,9 @@
)
from pyiceberg.utils.config import Config, merge_config

if TYPE_CHECKING:
 import pyarrow as pa

logger = logging.getLogger(__name__)

_ENV_CONFIG = Config()
Expand DownExpand Up@@ -288,7 +292,7 @@ def _load_file_io(self, properties: Properties = EMPTY_DICT, location: Optional[
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand DownExpand Up@@ -512,6 +516,22 @@ def _check_for_overlap(removals: Optional[Set[str]], updates: Properties) -> Non
 if overlap:
 raise ValueError(f"Updates and deletes have an overlap: {overlap}")

 @staticmethod
 def _convert_schema_if_needed(schema: Union[Schema, "pa.Schema"]) -> Schema:
 if isinstance(schema, Schema):
 return schema
 try:
Comment thread
HonahX marked this conversation as resolved.

 import pyarrow as pa

 from pyiceberg.io.pyarrow import _ConvertToIcebergWithoutIDs, visit_pyarrow

 if isinstance(schema, pa.Schema):
 schema: Schema = visit_pyarrow(schema, _ConvertToIcebergWithoutIDs()) # type: ignore
 return schema
 except ModuleNotFoundError:
 pass
 raise ValueError(f"{type(schema)=}, but it must be pyiceberg.schema.Schema or pyarrow.Schema")

 def _resolve_table_location(self, location: Optional[str], database_name: str, table_name: str) -> str:
 if not location:
 return self._get_default_warehouse_location(database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/dynamodb.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@
import uuid
from time import time
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -57,6 +58,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa

DYNAMODB_CLIENT = "dynamodb"

DYNAMODB_COL_IDENTIFIER = "identifier"
Expand DownExpand Up@@ -127,7 +131,7 @@ def _dynamodb_table_exists(self) -> bool:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -152,6 +156,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/glue.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@


from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -88,6 +89,9 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa

# If Glue should skip archiving an old table version when creating a new version in a commit. By
# default, Glue archives all old table versions after an UpdateTable call, but Glue has a default
# max number of archived table versions (can be increased). So for streaming use case with lots
Expand DownExpand Up@@ -329,7 +333,7 @@ def _get_glue_table(self, database_name: str, table_name: str) -> TableTypeDef:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -354,6 +358,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

9 changes: 8 additions & 1 deletion
pyiceberg/catalog/hive.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -18,6 +18,7 @@
import time
from types import TracebackType
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -91,6 +92,10 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa


# Replace by visitor
hive_types = {
 BooleanType: "boolean",
Expand DownExpand Up@@ -250,7 +255,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable, io: FileIO) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -273,6 +278,8 @@ def create_table(
 AlreadyExistsError: If a table with the name already exists.
 ValueError: If the identifier is invalid.
 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 properties = {**DEFAULT_PROPERTIES, **properties}
 database_name, table_name = self.identifier_to_database_and_table(identifier)
 current_time_millis = int(time.time() * 1000)
Expand Down

6 changes: 5 additions & 1 deletion
pyiceberg/catalog/noop.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand All@@ -33,12 +34,15 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties

if TYPE_CHECKING:
 import pyarrow as pa


class NoopCatalog(Catalog):
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/rest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
from json import JSONDecodeError
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -68,6 +69,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel

if TYPE_CHECKING:
 import pyarrow as pa

ICEBERG_REST_SPEC_VERSION = "0.14.1"


Expand DownExpand Up@@ -437,12 +441,14 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
 properties: Properties = EMPTY_DICT,
 ) -> Table:
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 namespace_and_table = self._split_identifier_for_path(identifier)
 request = CreateTableRequest(
 name=namespace_and_table["table"],
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.

from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand DownExpand Up@@ -65,6 +66,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa


class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase):
 pass
Expand DownExpand Up@@ -140,7 +144,7 @@ def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -165,6 +169,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)
 if not self._namespace_exists(database_name):
 raise NoSuchNamespaceError(f"Namespace does not exist: {database_name}")
Expand Down

25 changes: 20 additions & 5 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,7 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
Expand All@@ -34,7 +35,6 @@
from dataclasses import dataclass
from enum import Enum
from functools import lru_cache, singledispatch
from itertools import chain
from typing import (
 TYPE_CHECKING,
 Any,
Expand DownExpand Up@@ -631,7 +631,7 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:
 if len(positional_deletes) == 1:
 all_chunks = positional_deletes[0]
 else:
 all_chunks = pa.chunked_array(chain(*[arr.chunks for arr in positional_deletes]))
 all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
 return np.setdiff1d(np.arange(rows), all_chunks, assume_unique=False)


Expand DownExpand Up@@ -906,6 +906,21 @@ def after_map_value(self, element: pa.Field) -> None:
 self._field_names.pop()


class _ConvertToIcebergWithoutIDs(_ConvertToIceberg):
 """
 Converts PyArrowSchema to Iceberg Schema with all -1 ids.

 The schema generated through this visitor should always be
 used in conjunction with `new_table_metadata` function to
 assign new field ids in order. This is currently used only
 when creating an Iceberg Schema from a PyArrow schema when
 creating a new Iceberg table.
 """

 def _field_id(self, field: pa.Field) -> int:
 return -1


def _task_to_table(
 fs: FileSystem,
 task: FileScanTask,
Expand DownExpand Up@@ -993,7 +1008,7 @@ def _task_to_table(

def _read_all_delete_files(fs: FileSystem, tasks: Iterable[FileScanTask]) -> Dict[str, List[ChunkedArray]]:
 deletes_per_file: Dict[str, List[ChunkedArray]] = {}
 unique_deletes = set(chain.from_iterable([task.delete_files for task in tasks]))
 unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks]))
 if len(unique_deletes) > 0:
 executor = ExecutorFactory.get_or_create()
 deletes_per_files: Iterator[Dict[str, ChunkedArray]] = executor.map(
Expand DownExpand Up@@ -1399,7 +1414,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[StatisticsColl
 def struct(
 self, struct: StructType, field_results: List[Callable[[], List[StatisticsCollector]]]
 ) -> List[StatisticsCollector]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[StatisticsCollector]]) -> List[StatisticsCollector]:
 self._field_id = field.field_id
Expand DownExpand Up@@ -1491,7 +1506,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[ID2ParquetPath
 return struct_result()

 def struct(self, struct: StructType, field_results: List[Callable[[], List[ID2ParquetPath]]]) -> List[ID2ParquetPath]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[ID2ParquetPath]]) -> List[ID2ParquetPath]:
 self._field_id = field.field_id
Expand Down
Loading

, '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
create_table with a PyArrow Schema#305

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

HonahX
merged 9 commits into

from

Jan 30, 2024
 Merged

create_table with a PyArrow Schema
#305
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


9 commits


Select commit
Hold shift + click to select a range
8cd0e52

create_table from pyarrow schema


sungwy Jan 25, 2024
3d0445b

incorporate review comments


sungwy Jan 25, 2024
6ea0892

docs


sungwy Jan 25, 2024
7ba0fd1

Take suggestion


sungwy Jan 27, 2024
39a97a2

Take suggestion


sungwy Jan 27, 2024
60ac8f8

take nit


sungwy Jan 27, 2024
b917afd

more nit


sungwy Jan 27, 2024
e4e9f9b

refactoring


sungwy Jan 29, 2024
c40c553

adopt review suggestions


sungwy Jan 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


19 changes: 19 additions & 0 deletions
mkdocs/docs/api.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -146,6 +146,25 @@ catalog.create_table(
)
```

To create a table using a pyarrow schema:

```python
import pyarrow as pa

schema = pa.schema(
 [
 pa.field("foo", pa.string(), nullable=True),
 pa.field("bar", pa.int32(), nullable=False),
 pa.field("baz", pa.bool_(), nullable=True),
 ]
)

catalog.create_table(
 identifier="docs_example.bids",
 schema=schema,
)
```

## Load a table

### Catalog table
Expand Down

22 changes: 21 additions & 1 deletion
pyiceberg/catalog/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,6 +24,7 @@
from dataclasses import dataclass
from enum import Enum
from typing import (
 TYPE_CHECKING,
 Callable,
 Dict,
 List,
Expand DownExpand Up@@ -56,6 +57,9 @@
)
from pyiceberg.utils.config import Config, merge_config

if TYPE_CHECKING:
 import pyarrow as pa

logger = logging.getLogger(__name__)

_ENV_CONFIG = Config()
Expand DownExpand Up@@ -288,7 +292,7 @@ def _load_file_io(self, properties: Properties = EMPTY_DICT, location: Optional[
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand DownExpand Up@@ -512,6 +516,22 @@ def _check_for_overlap(removals: Optional[Set[str]], updates: Properties) -> Non
 if overlap:
 raise ValueError(f"Updates and deletes have an overlap: {overlap}")

 @staticmethod
 def _convert_schema_if_needed(schema: Union[Schema, "pa.Schema"]) -> Schema:
 if isinstance(schema, Schema):
 return schema
 try:
Comment thread
HonahX marked this conversation as resolved.

 import pyarrow as pa

 from pyiceberg.io.pyarrow import _ConvertToIcebergWithoutIDs, visit_pyarrow

 if isinstance(schema, pa.Schema):
 schema: Schema = visit_pyarrow(schema, _ConvertToIcebergWithoutIDs()) # type: ignore
 return schema
 except ModuleNotFoundError:
 pass
 raise ValueError(f"{type(schema)=}, but it must be pyiceberg.schema.Schema or pyarrow.Schema")

 def _resolve_table_location(self, location: Optional[str], database_name: str, table_name: str) -> str:
 if not location:
 return self._get_default_warehouse_location(database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/dynamodb.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@
import uuid
from time import time
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -57,6 +58,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa

DYNAMODB_CLIENT = "dynamodb"

DYNAMODB_COL_IDENTIFIER = "identifier"
Expand DownExpand Up@@ -127,7 +131,7 @@ def _dynamodb_table_exists(self) -> bool:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -152,6 +156,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/glue.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@


from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -88,6 +89,9 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa

# If Glue should skip archiving an old table version when creating a new version in a commit. By
# default, Glue archives all old table versions after an UpdateTable call, but Glue has a default
# max number of archived table versions (can be increased). So for streaming use case with lots
Expand DownExpand Up@@ -329,7 +333,7 @@ def _get_glue_table(self, database_name: str, table_name: str) -> TableTypeDef:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -354,6 +358,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

9 changes: 8 additions & 1 deletion
pyiceberg/catalog/hive.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -18,6 +18,7 @@
import time
from types import TracebackType
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -91,6 +92,10 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa


# Replace by visitor
hive_types = {
 BooleanType: "boolean",
Expand DownExpand Up@@ -250,7 +255,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable, io: FileIO) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -273,6 +278,8 @@ def create_table(
 AlreadyExistsError: If a table with the name already exists.
 ValueError: If the identifier is invalid.
 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 properties = {**DEFAULT_PROPERTIES, **properties}
 database_name, table_name = self.identifier_to_database_and_table(identifier)
 current_time_millis = int(time.time() * 1000)
Expand Down

6 changes: 5 additions & 1 deletion
pyiceberg/catalog/noop.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand All@@ -33,12 +34,15 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties

if TYPE_CHECKING:
 import pyarrow as pa


class NoopCatalog(Catalog):
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/rest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
from json import JSONDecodeError
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -68,6 +69,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel

if TYPE_CHECKING:
 import pyarrow as pa

ICEBERG_REST_SPEC_VERSION = "0.14.1"


Expand DownExpand Up@@ -437,12 +441,14 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
 properties: Properties = EMPTY_DICT,
 ) -> Table:
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 namespace_and_table = self._split_identifier_for_path(identifier)
 request = CreateTableRequest(
 name=namespace_and_table["table"],
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.

from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand DownExpand Up@@ -65,6 +66,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa


class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase):
 pass
Expand DownExpand Up@@ -140,7 +144,7 @@ def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -165,6 +169,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)
 if not self._namespace_exists(database_name):
 raise NoSuchNamespaceError(f"Namespace does not exist: {database_name}")
Expand Down

25 changes: 20 additions & 5 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,7 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
Expand All@@ -34,7 +35,6 @@
from dataclasses import dataclass
from enum import Enum
from functools import lru_cache, singledispatch
from itertools import chain
from typing import (
 TYPE_CHECKING,
 Any,
Expand DownExpand Up@@ -631,7 +631,7 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:
 if len(positional_deletes) == 1:
 all_chunks = positional_deletes[0]
 else:
 all_chunks = pa.chunked_array(chain(*[arr.chunks for arr in positional_deletes]))
 all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
 return np.setdiff1d(np.arange(rows), all_chunks, assume_unique=False)


Expand DownExpand Up@@ -906,6 +906,21 @@ def after_map_value(self, element: pa.Field) -> None:
 self._field_names.pop()


class _ConvertToIcebergWithoutIDs(_ConvertToIceberg):
 """
 Converts PyArrowSchema to Iceberg Schema with all -1 ids.

 The schema generated through this visitor should always be
 used in conjunction with `new_table_metadata` function to
 assign new field ids in order. This is currently used only
 when creating an Iceberg Schema from a PyArrow schema when
 creating a new Iceberg table.
 """

 def _field_id(self, field: pa.Field) -> int:
 return -1


def _task_to_table(
 fs: FileSystem,
 task: FileScanTask,
Expand DownExpand Up@@ -993,7 +1008,7 @@ def _task_to_table(

def _read_all_delete_files(fs: FileSystem, tasks: Iterable[FileScanTask]) -> Dict[str, List[ChunkedArray]]:
 deletes_per_file: Dict[str, List[ChunkedArray]] = {}
 unique_deletes = set(chain.from_iterable([task.delete_files for task in tasks]))
 unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks]))
 if len(unique_deletes) > 0:
 executor = ExecutorFactory.get_or_create()
 deletes_per_files: Iterator[Dict[str, ChunkedArray]] = executor.map(
Expand DownExpand Up@@ -1399,7 +1414,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[StatisticsColl
 def struct(
 self, struct: StructType, field_results: List[Callable[[], List[StatisticsCollector]]]
 ) -> List[StatisticsCollector]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[StatisticsCollector]]) -> List[StatisticsCollector]:
 self._field_id = field.field_id
Expand DownExpand Up@@ -1491,7 +1506,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[ID2ParquetPath
 return struct_result()

 def struct(self, struct: StructType, field_results: List[Callable[[], List[ID2ParquetPath]]]) -> List[ID2ParquetPath]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[ID2ParquetPath]]) -> List[ID2ParquetPath]:
 self._field_id = field.field_id
Expand Down
Loading

, '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
create_table with a PyArrow Schema#305

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

HonahX
merged 9 commits into

from

Jan 30, 2024
 Merged

create_table with a PyArrow Schema
#305
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


9 commits


Select commit
Hold shift + click to select a range
8cd0e52

create_table from pyarrow schema


sungwy Jan 25, 2024
3d0445b

incorporate review comments


sungwy Jan 25, 2024
6ea0892

docs


sungwy Jan 25, 2024
7ba0fd1

Take suggestion


sungwy Jan 27, 2024
39a97a2

Take suggestion


sungwy Jan 27, 2024
60ac8f8

take nit


sungwy Jan 27, 2024
b917afd

more nit


sungwy Jan 27, 2024
e4e9f9b

refactoring


sungwy Jan 29, 2024
c40c553

adopt review suggestions


sungwy Jan 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


19 changes: 19 additions & 0 deletions
mkdocs/docs/api.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -146,6 +146,25 @@ catalog.create_table(
)
```

To create a table using a pyarrow schema:

```python
import pyarrow as pa

schema = pa.schema(
 [
 pa.field("foo", pa.string(), nullable=True),
 pa.field("bar", pa.int32(), nullable=False),
 pa.field("baz", pa.bool_(), nullable=True),
 ]
)

catalog.create_table(
 identifier="docs_example.bids",
 schema=schema,
)
```

## Load a table

### Catalog table
Expand Down

22 changes: 21 additions & 1 deletion
pyiceberg/catalog/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,6 +24,7 @@
from dataclasses import dataclass
from enum import Enum
from typing import (
 TYPE_CHECKING,
 Callable,
 Dict,
 List,
Expand DownExpand Up@@ -56,6 +57,9 @@
)
from pyiceberg.utils.config import Config, merge_config

if TYPE_CHECKING:
 import pyarrow as pa

logger = logging.getLogger(__name__)

_ENV_CONFIG = Config()
Expand DownExpand Up@@ -288,7 +292,7 @@ def _load_file_io(self, properties: Properties = EMPTY_DICT, location: Optional[
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand DownExpand Up@@ -512,6 +516,22 @@ def _check_for_overlap(removals: Optional[Set[str]], updates: Properties) -> Non
 if overlap:
 raise ValueError(f"Updates and deletes have an overlap: {overlap}")

 @staticmethod
 def _convert_schema_if_needed(schema: Union[Schema, "pa.Schema"]) -> Schema:
 if isinstance(schema, Schema):
 return schema
 try:
Comment thread
HonahX marked this conversation as resolved.

 import pyarrow as pa

 from pyiceberg.io.pyarrow import _ConvertToIcebergWithoutIDs, visit_pyarrow

 if isinstance(schema, pa.Schema):
 schema: Schema = visit_pyarrow(schema, _ConvertToIcebergWithoutIDs()) # type: ignore
 return schema
 except ModuleNotFoundError:
 pass
 raise ValueError(f"{type(schema)=}, but it must be pyiceberg.schema.Schema or pyarrow.Schema")

 def _resolve_table_location(self, location: Optional[str], database_name: str, table_name: str) -> str:
 if not location:
 return self._get_default_warehouse_location(database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/dynamodb.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@
import uuid
from time import time
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -57,6 +58,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa

DYNAMODB_CLIENT = "dynamodb"

DYNAMODB_COL_IDENTIFIER = "identifier"
Expand DownExpand Up@@ -127,7 +131,7 @@ def _dynamodb_table_exists(self) -> bool:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -152,6 +156,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/glue.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@


from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -88,6 +89,9 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa

# If Glue should skip archiving an old table version when creating a new version in a commit. By
# default, Glue archives all old table versions after an UpdateTable call, but Glue has a default
# max number of archived table versions (can be increased). So for streaming use case with lots
Expand DownExpand Up@@ -329,7 +333,7 @@ def _get_glue_table(self, database_name: str, table_name: str) -> TableTypeDef:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -354,6 +358,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

9 changes: 8 additions & 1 deletion
pyiceberg/catalog/hive.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -18,6 +18,7 @@
import time
from types import TracebackType
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -91,6 +92,10 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa


# Replace by visitor
hive_types = {
 BooleanType: "boolean",
Expand DownExpand Up@@ -250,7 +255,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable, io: FileIO) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -273,6 +278,8 @@ def create_table(
 AlreadyExistsError: If a table with the name already exists.
 ValueError: If the identifier is invalid.
 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 properties = {**DEFAULT_PROPERTIES, **properties}
 database_name, table_name = self.identifier_to_database_and_table(identifier)
 current_time_millis = int(time.time() * 1000)
Expand Down

6 changes: 5 additions & 1 deletion
pyiceberg/catalog/noop.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand All@@ -33,12 +34,15 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties

if TYPE_CHECKING:
 import pyarrow as pa


class NoopCatalog(Catalog):
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/rest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
from json import JSONDecodeError
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -68,6 +69,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel

if TYPE_CHECKING:
 import pyarrow as pa

ICEBERG_REST_SPEC_VERSION = "0.14.1"


Expand DownExpand Up@@ -437,12 +441,14 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
 properties: Properties = EMPTY_DICT,
 ) -> Table:
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 namespace_and_table = self._split_identifier_for_path(identifier)
 request = CreateTableRequest(
 name=namespace_and_table["table"],
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.

from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand DownExpand Up@@ -65,6 +66,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa


class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase):
 pass
Expand DownExpand Up@@ -140,7 +144,7 @@ def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -165,6 +169,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)
 if not self._namespace_exists(database_name):
 raise NoSuchNamespaceError(f"Namespace does not exist: {database_name}")
Expand Down

25 changes: 20 additions & 5 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,7 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
Expand All@@ -34,7 +35,6 @@
from dataclasses import dataclass
from enum import Enum
from functools import lru_cache, singledispatch
from itertools import chain
from typing import (
 TYPE_CHECKING,
 Any,
Expand DownExpand Up@@ -631,7 +631,7 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:
 if len(positional_deletes) == 1:
 all_chunks = positional_deletes[0]
 else:
 all_chunks = pa.chunked_array(chain(*[arr.chunks for arr in positional_deletes]))
 all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
 return np.setdiff1d(np.arange(rows), all_chunks, assume_unique=False)


Expand DownExpand Up@@ -906,6 +906,21 @@ def after_map_value(self, element: pa.Field) -> None:
 self._field_names.pop()


class _ConvertToIcebergWithoutIDs(_ConvertToIceberg):
 """
 Converts PyArrowSchema to Iceberg Schema with all -1 ids.

 The schema generated through this visitor should always be
 used in conjunction with `new_table_metadata` function to
 assign new field ids in order. This is currently used only
 when creating an Iceberg Schema from a PyArrow schema when
 creating a new Iceberg table.
 """

 def _field_id(self, field: pa.Field) -> int:
 return -1


def _task_to_table(
 fs: FileSystem,
 task: FileScanTask,
Expand DownExpand Up@@ -993,7 +1008,7 @@ def _task_to_table(

def _read_all_delete_files(fs: FileSystem, tasks: Iterable[FileScanTask]) -> Dict[str, List[ChunkedArray]]:
 deletes_per_file: Dict[str, List[ChunkedArray]] = {}
 unique_deletes = set(chain.from_iterable([task.delete_files for task in tasks]))
 unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks]))
 if len(unique_deletes) > 0:
 executor = ExecutorFactory.get_or_create()
 deletes_per_files: Iterator[Dict[str, ChunkedArray]] = executor.map(
Expand DownExpand Up@@ -1399,7 +1414,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[StatisticsColl
 def struct(
 self, struct: StructType, field_results: List[Callable[[], List[StatisticsCollector]]]
 ) -> List[StatisticsCollector]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[StatisticsCollector]]) -> List[StatisticsCollector]:
 self._field_id = field.field_id
Expand DownExpand Up@@ -1491,7 +1506,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[ID2ParquetPath
 return struct_result()

 def struct(self, struct: StructType, field_results: List[Callable[[], List[ID2ParquetPath]]]) -> List[ID2ParquetPath]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[ID2ParquetPath]]) -> List[ID2ParquetPath]:
 self._field_id = field.field_id
Expand Down
Loading

, '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
create_table with a PyArrow Schema#305

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

HonahX
merged 9 commits into

from

Jan 30, 2024
 Merged

create_table with a PyArrow Schema
#305
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


9 commits


Select commit
Hold shift + click to select a range
8cd0e52

create_table from pyarrow schema


sungwy Jan 25, 2024
3d0445b

incorporate review comments


sungwy Jan 25, 2024
6ea0892

docs


sungwy Jan 25, 2024
7ba0fd1

Take suggestion


sungwy Jan 27, 2024
39a97a2

Take suggestion


sungwy Jan 27, 2024
60ac8f8

take nit


sungwy Jan 27, 2024
b917afd

more nit


sungwy Jan 27, 2024
e4e9f9b

refactoring


sungwy Jan 29, 2024
c40c553

adopt review suggestions


sungwy Jan 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


19 changes: 19 additions & 0 deletions
mkdocs/docs/api.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -146,6 +146,25 @@ catalog.create_table(
)
```

To create a table using a pyarrow schema:

```python
import pyarrow as pa

schema = pa.schema(
 [
 pa.field("foo", pa.string(), nullable=True),
 pa.field("bar", pa.int32(), nullable=False),
 pa.field("baz", pa.bool_(), nullable=True),
 ]
)

catalog.create_table(
 identifier="docs_example.bids",
 schema=schema,
)
```

## Load a table

### Catalog table
Expand Down

22 changes: 21 additions & 1 deletion
pyiceberg/catalog/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,6 +24,7 @@
from dataclasses import dataclass
from enum import Enum
from typing import (
 TYPE_CHECKING,
 Callable,
 Dict,
 List,
Expand DownExpand Up@@ -56,6 +57,9 @@
)
from pyiceberg.utils.config import Config, merge_config

if TYPE_CHECKING:
 import pyarrow as pa

logger = logging.getLogger(__name__)

_ENV_CONFIG = Config()
Expand DownExpand Up@@ -288,7 +292,7 @@ def _load_file_io(self, properties: Properties = EMPTY_DICT, location: Optional[
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand DownExpand Up@@ -512,6 +516,22 @@ def _check_for_overlap(removals: Optional[Set[str]], updates: Properties) -> Non
 if overlap:
 raise ValueError(f"Updates and deletes have an overlap: {overlap}")

 @staticmethod
 def _convert_schema_if_needed(schema: Union[Schema, "pa.Schema"]) -> Schema:
 if isinstance(schema, Schema):
 return schema
 try:
Comment thread
HonahX marked this conversation as resolved.

 import pyarrow as pa

 from pyiceberg.io.pyarrow import _ConvertToIcebergWithoutIDs, visit_pyarrow

 if isinstance(schema, pa.Schema):
 schema: Schema = visit_pyarrow(schema, _ConvertToIcebergWithoutIDs()) # type: ignore
 return schema
 except ModuleNotFoundError:
 pass
 raise ValueError(f"{type(schema)=}, but it must be pyiceberg.schema.Schema or pyarrow.Schema")

 def _resolve_table_location(self, location: Optional[str], database_name: str, table_name: str) -> str:
 if not location:
 return self._get_default_warehouse_location(database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/dynamodb.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@
import uuid
from time import time
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -57,6 +58,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa

DYNAMODB_CLIENT = "dynamodb"

DYNAMODB_COL_IDENTIFIER = "identifier"
Expand DownExpand Up@@ -127,7 +131,7 @@ def _dynamodb_table_exists(self) -> bool:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -152,6 +156,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/glue.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@


from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -88,6 +89,9 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa

# If Glue should skip archiving an old table version when creating a new version in a commit. By
# default, Glue archives all old table versions after an UpdateTable call, but Glue has a default
# max number of archived table versions (can be increased). So for streaming use case with lots
Expand DownExpand Up@@ -329,7 +333,7 @@ def _get_glue_table(self, database_name: str, table_name: str) -> TableTypeDef:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -354,6 +358,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

9 changes: 8 additions & 1 deletion
pyiceberg/catalog/hive.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -18,6 +18,7 @@
import time
from types import TracebackType
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -91,6 +92,10 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa


# Replace by visitor
hive_types = {
 BooleanType: "boolean",
Expand DownExpand Up@@ -250,7 +255,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable, io: FileIO) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -273,6 +278,8 @@ def create_table(
 AlreadyExistsError: If a table with the name already exists.
 ValueError: If the identifier is invalid.
 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 properties = {**DEFAULT_PROPERTIES, **properties}
 database_name, table_name = self.identifier_to_database_and_table(identifier)
 current_time_millis = int(time.time() * 1000)
Expand Down

6 changes: 5 additions & 1 deletion
pyiceberg/catalog/noop.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand All@@ -33,12 +34,15 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties

if TYPE_CHECKING:
 import pyarrow as pa


class NoopCatalog(Catalog):
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/rest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
from json import JSONDecodeError
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -68,6 +69,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel

if TYPE_CHECKING:
 import pyarrow as pa

ICEBERG_REST_SPEC_VERSION = "0.14.1"


Expand DownExpand Up@@ -437,12 +441,14 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
 properties: Properties = EMPTY_DICT,
 ) -> Table:
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 namespace_and_table = self._split_identifier_for_path(identifier)
 request = CreateTableRequest(
 name=namespace_and_table["table"],
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.

from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand DownExpand Up@@ -65,6 +66,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa


class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase):
 pass
Expand DownExpand Up@@ -140,7 +144,7 @@ def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -165,6 +169,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)
 if not self._namespace_exists(database_name):
 raise NoSuchNamespaceError(f"Namespace does not exist: {database_name}")
Expand Down

25 changes: 20 additions & 5 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,7 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
Expand All@@ -34,7 +35,6 @@
from dataclasses import dataclass
from enum import Enum
from functools import lru_cache, singledispatch
from itertools import chain
from typing import (
 TYPE_CHECKING,
 Any,
Expand DownExpand Up@@ -631,7 +631,7 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:
 if len(positional_deletes) == 1:
 all_chunks = positional_deletes[0]
 else:
 all_chunks = pa.chunked_array(chain(*[arr.chunks for arr in positional_deletes]))
 all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
 return np.setdiff1d(np.arange(rows), all_chunks, assume_unique=False)


Expand DownExpand Up@@ -906,6 +906,21 @@ def after_map_value(self, element: pa.Field) -> None:
 self._field_names.pop()


class _ConvertToIcebergWithoutIDs(_ConvertToIceberg):
 """
 Converts PyArrowSchema to Iceberg Schema with all -1 ids.

 The schema generated through this visitor should always be
 used in conjunction with `new_table_metadata` function to
 assign new field ids in order. This is currently used only
 when creating an Iceberg Schema from a PyArrow schema when
 creating a new Iceberg table.
 """

 def _field_id(self, field: pa.Field) -> int:
 return -1


def _task_to_table(
 fs: FileSystem,
 task: FileScanTask,
Expand DownExpand Up@@ -993,7 +1008,7 @@ def _task_to_table(

def _read_all_delete_files(fs: FileSystem, tasks: Iterable[FileScanTask]) -> Dict[str, List[ChunkedArray]]:
 deletes_per_file: Dict[str, List[ChunkedArray]] = {}
 unique_deletes = set(chain.from_iterable([task.delete_files for task in tasks]))
 unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks]))
 if len(unique_deletes) > 0:
 executor = ExecutorFactory.get_or_create()
 deletes_per_files: Iterator[Dict[str, ChunkedArray]] = executor.map(
Expand DownExpand Up@@ -1399,7 +1414,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[StatisticsColl
 def struct(
 self, struct: StructType, field_results: List[Callable[[], List[StatisticsCollector]]]
 ) -> List[StatisticsCollector]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[StatisticsCollector]]) -> List[StatisticsCollector]:
 self._field_id = field.field_id
Expand DownExpand Up@@ -1491,7 +1506,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[ID2ParquetPath
 return struct_result()

 def struct(self, struct: StructType, field_results: List[Callable[[], List[ID2ParquetPath]]]) -> List[ID2ParquetPath]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[ID2ParquetPath]]) -> List[ID2ParquetPath]:
 self._field_id = field.field_id
Expand Down
Loading

, '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
create_table with a PyArrow Schema#305

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

HonahX
merged 9 commits into

from

Jan 30, 2024
 Merged

create_table with a PyArrow Schema
#305
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


9 commits


Select commit
Hold shift + click to select a range
8cd0e52

create_table from pyarrow schema


sungwy Jan 25, 2024
3d0445b

incorporate review comments


sungwy Jan 25, 2024
6ea0892

docs


sungwy Jan 25, 2024
7ba0fd1

Take suggestion


sungwy Jan 27, 2024
39a97a2

Take suggestion


sungwy Jan 27, 2024
60ac8f8

take nit


sungwy Jan 27, 2024
b917afd

more nit


sungwy Jan 27, 2024
e4e9f9b

refactoring


sungwy Jan 29, 2024
c40c553

adopt review suggestions


sungwy Jan 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


19 changes: 19 additions & 0 deletions
mkdocs/docs/api.md

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -146,6 +146,25 @@ catalog.create_table(
)
```

To create a table using a pyarrow schema:

```python
import pyarrow as pa

schema = pa.schema(
 [
 pa.field("foo", pa.string(), nullable=True),
 pa.field("bar", pa.int32(), nullable=False),
 pa.field("baz", pa.bool_(), nullable=True),
 ]
)

catalog.create_table(
 identifier="docs_example.bids",
 schema=schema,
)
```

## Load a table

### Catalog table
Expand Down

22 changes: 21 additions & 1 deletion
pyiceberg/catalog/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -24,6 +24,7 @@
from dataclasses import dataclass
from enum import Enum
from typing import (
 TYPE_CHECKING,
 Callable,
 Dict,
 List,
Expand DownExpand Up@@ -56,6 +57,9 @@
)
from pyiceberg.utils.config import Config, merge_config

if TYPE_CHECKING:
 import pyarrow as pa

logger = logging.getLogger(__name__)

_ENV_CONFIG = Config()
Expand DownExpand Up@@ -288,7 +292,7 @@ def _load_file_io(self, properties: Properties = EMPTY_DICT, location: Optional[
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand DownExpand Up@@ -512,6 +516,22 @@ def _check_for_overlap(removals: Optional[Set[str]], updates: Properties) -> Non
 if overlap:
 raise ValueError(f"Updates and deletes have an overlap: {overlap}")

 @staticmethod
 def _convert_schema_if_needed(schema: Union[Schema, "pa.Schema"]) -> Schema:
 if isinstance(schema, Schema):
 return schema
 try:
Comment thread
HonahX marked this conversation as resolved.

 import pyarrow as pa

 from pyiceberg.io.pyarrow import _ConvertToIcebergWithoutIDs, visit_pyarrow

 if isinstance(schema, pa.Schema):
 schema: Schema = visit_pyarrow(schema, _ConvertToIcebergWithoutIDs()) # type: ignore
 return schema
 except ModuleNotFoundError:
 pass
 raise ValueError(f"{type(schema)=}, but it must be pyiceberg.schema.Schema or pyarrow.Schema")

 def _resolve_table_location(self, location: Optional[str], database_name: str, table_name: str) -> str:
 if not location:
 return self._get_default_warehouse_location(database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/dynamodb.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@
import uuid
from time import time
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -57,6 +58,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa

DYNAMODB_CLIENT = "dynamodb"

DYNAMODB_COL_IDENTIFIER = "identifier"
Expand DownExpand Up@@ -127,7 +131,7 @@ def _dynamodb_table_exists(self) -> bool:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -152,6 +156,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/glue.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,7 @@


from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -88,6 +89,9 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa

# If Glue should skip archiving an old table version when creating a new version in a commit. By
# default, Glue archives all old table versions after an UpdateTable call, but Glue has a default
# max number of archived table versions (can be increased). So for streaming use case with lots
Expand DownExpand Up@@ -329,7 +333,7 @@ def _get_glue_table(self, database_name: str, table_name: str) -> TableTypeDef:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -354,6 +358,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)

 location = self._resolve_table_location(location, database_name, table_name)
Expand Down

9 changes: 8 additions & 1 deletion
pyiceberg/catalog/hive.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -18,6 +18,7 @@
import time
from types import TracebackType
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -91,6 +92,10 @@
 UUIDType,
)

if TYPE_CHECKING:
 import pyarrow as pa


# Replace by visitor
hive_types = {
 BooleanType: "boolean",
Expand DownExpand Up@@ -250,7 +255,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable, io: FileIO) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -273,6 +278,8 @@ def create_table(
 AlreadyExistsError: If a table with the name already exists.
 ValueError: If the identifier is invalid.
 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 properties = {**DEFAULT_PROPERTIES, **properties}
 database_name, table_name = self.identifier_to_database_and_table(identifier)
 current_time_millis = int(time.time() * 1000)
Expand Down

6 changes: 5 additions & 1 deletion
pyiceberg/catalog/noop.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand All@@ -33,12 +34,15 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties

if TYPE_CHECKING:
 import pyarrow as pa


class NoopCatalog(Catalog):
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/rest.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.
from json import JSONDecodeError
from typing import (
 TYPE_CHECKING,
 Any,
 Dict,
 List,
Expand DownExpand Up@@ -68,6 +69,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel

if TYPE_CHECKING:
 import pyarrow as pa

ICEBERG_REST_SPEC_VERSION = "0.14.1"


Expand DownExpand Up@@ -437,12 +441,14 @@ def _response_to_table(self, identifier_tuple: Tuple[str, ...], table_response:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
 properties: Properties = EMPTY_DICT,
 ) -> Table:
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 namespace_and_table = self._split_identifier_for_path(identifier)
 request = CreateTableRequest(
 name=namespace_and_table["table"],
Expand Down

8 changes: 7 additions & 1 deletion
pyiceberg/catalog/sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -16,6 +16,7 @@
# under the License.

from typing import (
 TYPE_CHECKING,
 List,
 Optional,
 Set,
Expand DownExpand Up@@ -65,6 +66,9 @@
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT

if TYPE_CHECKING:
 import pyarrow as pa


class SqlCatalogBaseTable(MappedAsDataclass, DeclarativeBase):
 pass
Expand DownExpand Up@@ -140,7 +144,7 @@ def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table:
 def create_table(
 self,
 identifier: Union[str, Identifier],
 schema: Schema,
 schema: Union[Schema, "pa.Schema"],
 location: Optional[str] = None,
 partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
 sort_order: SortOrder = UNSORTED_SORT_ORDER,
Expand All@@ -165,6 +169,8 @@ def create_table(
 ValueError: If the identifier is invalid, or no path is given to store metadata.

 """
 schema: Schema = self._convert_schema_if_needed(schema) # type: ignore

 database_name, table_name = self.identifier_to_database_and_table(identifier)
 if not self._namespace_exists(database_name):
 raise NoSuchNamespaceError(f"Namespace does not exist: {database_name}")
Expand Down

25 changes: 20 additions & 5 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -26,6 +26,7 @@
from __future__ import annotations

import concurrent.futures
import itertools
import logging
import os
import re
Expand All@@ -34,7 +35,6 @@
from dataclasses import dataclass
from enum import Enum
from functools import lru_cache, singledispatch
from itertools import chain
from typing import (
 TYPE_CHECKING,
 Any,
Expand DownExpand Up@@ -631,7 +631,7 @@ def _combine_positional_deletes(positional_deletes: List[pa.ChunkedArray], rows:
 if len(positional_deletes) == 1:
 all_chunks = positional_deletes[0]
 else:
 all_chunks = pa.chunked_array(chain(*[arr.chunks for arr in positional_deletes]))
 all_chunks = pa.chunked_array(itertools.chain(*[arr.chunks for arr in positional_deletes]))
 return np.setdiff1d(np.arange(rows), all_chunks, assume_unique=False)


Expand DownExpand Up@@ -906,6 +906,21 @@ def after_map_value(self, element: pa.Field) -> None:
 self._field_names.pop()


class _ConvertToIcebergWithoutIDs(_ConvertToIceberg):
 """
 Converts PyArrowSchema to Iceberg Schema with all -1 ids.

 The schema generated through this visitor should always be
 used in conjunction with `new_table_metadata` function to
 assign new field ids in order. This is currently used only
 when creating an Iceberg Schema from a PyArrow schema when
 creating a new Iceberg table.
 """

 def _field_id(self, field: pa.Field) -> int:
 return -1


def _task_to_table(
 fs: FileSystem,
 task: FileScanTask,
Expand DownExpand Up@@ -993,7 +1008,7 @@ def _task_to_table(

def _read_all_delete_files(fs: FileSystem, tasks: Iterable[FileScanTask]) -> Dict[str, List[ChunkedArray]]:
 deletes_per_file: Dict[str, List[ChunkedArray]] = {}
 unique_deletes = set(chain.from_iterable([task.delete_files for task in tasks]))
 unique_deletes = set(itertools.chain.from_iterable([task.delete_files for task in tasks]))
 if len(unique_deletes) > 0:
 executor = ExecutorFactory.get_or_create()
 deletes_per_files: Iterator[Dict[str, ChunkedArray]] = executor.map(
Expand DownExpand Up@@ -1399,7 +1414,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[StatisticsColl
 def struct(
 self, struct: StructType, field_results: List[Callable[[], List[StatisticsCollector]]]
 ) -> List[StatisticsCollector]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[StatisticsCollector]]) -> List[StatisticsCollector]:
 self._field_id = field.field_id
Expand DownExpand Up@@ -1491,7 +1506,7 @@ def schema(self, schema: Schema, struct_result: Callable[[], List[ID2ParquetPath
 return struct_result()

 def struct(self, struct: StructType, field_results: List[Callable[[], List[ID2ParquetPath]]]) -> List[ID2ParquetPath]:
 return list(chain(*[result() for result in field_results]))
 return list(itertools.chain(*[result() for result in field_results]))

 def field(self, field: NestedField, field_result: Callable[[], List[ID2ParquetPath]]) -> List[ID2ParquetPath]:
 self._field_id = field.field_id
Expand Down
Loading