Fix writing to local filesystem#301

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
Fokko merged 5 commits into from
Jan 29, 2024
Merged

Fix writing to local filesystem #301

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
369df6e
add a test for writing data to disk
kevinjqliu Jan 24, 2024
5751b80
create dir first
kevinjqliu Jan 24, 2024
4101dd1
add fsspec test
kevinjqliu Jan 26, 2024
57d68f2
make lint
kevinjqliu Jan 26, 2024
242e365
add fs tests
kevinjqliu Jan 26, 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
2 changes: 1 addition & 1 deletion pyiceberg/io/__init__.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -275,7 +275,7 @@ def delete(self, location: Union[str, InputFile, OutputFile]) -> None:
"s3a": [ARROW_FILE_IO, FSSPEC_FILE_IO],
"s3n": [ARROW_FILE_IO, FSSPEC_FILE_IO],
"gs": [ARROW_FILE_IO],
"file": [ARROW_FILE_IO],
"file": [ARROW_FILE_IO, FSSPEC_FILE_IO],
"hdfs": [ARROW_FILE_IO],
"abfs": [FSSPEC_FILE_IO],
"abfss": [FSSPEC_FILE_IO],
Expand Down
3 changes: 2 additions & 1 deletion pyiceberg/io/fsspec.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -99,7 +99,7 @@ def s3v4_rest_signer(properties: Properties, request: AWSRequest, **_: Any) -> A


def _file(_: Properties) -> LocalFileSystem:
return LocalFileSystem()
return LocalFileSystem(auto_mkdir=True)


def _s3(properties: Properties) -> AbstractFileSystem:
Expand DownExpand Up@@ -173,6 +173,7 @@ def _adlfs(properties: Properties) -> AbstractFileSystem:


SCHEME_TO_FS = {
"": _file,

@kevinjqliukevinjqliuJan 26, 2024

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pyarrow defaults scheme to file when no scheme is present.

iceberg-python/pyiceberg/io/pyarrow.py

Line 322 in 8464d71

return"file", uri.netloc, os.path.abspath(location)

we essentially do the same here

"file": _file,
"s3": _s3,
"s3a": _s3,
Expand Down
11 changes: 8 additions & 3 deletions pyiceberg/io/pyarrow.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -172,6 +172,13 @@
T = TypeVar("T")


class PyArrowLocalFileSystem(pyarrow.fs.LocalFileSystem):
def open_output_stream(self, path: str, *args: Any, **kwargs: Any) -> pyarrow.NativeFile:
# In LocalFileSystem, parent directories must be first created before opening an output stream
self.create_dir(os.path.dirname(path), recursive=True)
return super().open_output_stream(path, *args, **kwargs)


class PyArrowFile(InputFile, OutputFile):
"""A combined InputFile and OutputFile implementation that uses a pyarrow filesystem to generate pyarrow.lib.NativeFile instances.

Expand DownExpand Up@@ -378,9 +385,7 @@ def _initialize_fs(self, scheme: str, netloc: Optional[str] = None) -> FileSyste

return GcsFileSystem(**gcs_kwargs)
elif scheme == "file":
from pyarrow.fs import LocalFileSystem

return LocalFileSystem()
return PyArrowLocalFileSystem()
else:
raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")

Expand Down
62 changes: 60 additions & 2 deletions tests/catalog/test_sql.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -19,6 +19,7 @@
from pathlib import Path
from typing import Generator, List

import pyarrow as pa
import pytest
from pytest import TempPathFactory
from pytest_lazyfixture import lazy_fixture
Expand All@@ -35,7 +36,10 @@
NoSuchTableError,
TableAlreadyExistsError,
)
from pyiceberg.io import FSSPEC_FILE_IO, PY_IO_IMPL
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.schema import Schema
from pyiceberg.table.snapshots import Operation
from pyiceberg.table.sorting import (
NullOrder,
SortDirection,
Expand DownExpand Up@@ -80,7 +84,7 @@ def catalog_memory(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
props = {
"uri": "sqlite:////tmp/sql-catalog.db",
"uri": f"sqlite:////{warehouse}/sql-catalog.db",
"warehouse": f"file://{warehouse}",
}
catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -92,7 +96,7 @@ def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, None, None]:
props = {
"uri": "sqlite:////tmp/sql-catalog.db",
"uri": f"sqlite:////{warehouse}/sql-catalog.db",
"warehouse": f"file://{warehouse}",
}
catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -102,6 +106,19 @@ def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, No
catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite_fsspec(warehouse: Path) -> Generator[SqlCatalog, None, None]:
props = {
"uri": f"sqlite:////{warehouse}/sql-catalog.db",
"warehouse": f"file://{warehouse}",
PY_IO_IMPL: FSSPEC_FILE_IO,
}
catalog = SqlCatalog("test_sql_catalog", **props)
catalog.create_tables()
yield catalog
catalog.destroy_tables()


def test_creation_with_no_uri() -> None:
with pytest.raises(NoSuchPropertyException):
SqlCatalog("test_ddb_catalog", not_uri="unused")
Expand DownExpand Up@@ -722,6 +739,47 @@ def test_commit_table(catalog: SqlCatalog, table_schema_nested: Schema, random_i
assert new_schema.find_field("b").field_type == IntegerType()


@pytest.mark.parametrize(
'catalog',
[
lazy_fixture('catalog_memory'),
lazy_fixture('catalog_sqlite'),
lazy_fixture('catalog_sqlite_without_rowcount'),
lazy_fixture('catalog_sqlite_fsspec'),
],
)
def test_append_table(catalog: SqlCatalog, table_schema_simple: Schema, random_identifier: Identifier) -> None:
database_name, _table_name = random_identifier
catalog.create_namespace(database_name)
table = catalog.create_table(random_identifier, table_schema_simple)

df = pa.Table.from_pydict(
{
"foo": ["a"],
"bar": [1],
"baz": [True],
},
schema=schema_to_pyarrow(table_schema_simple),
)

table.append(df)

# new snapshot is written in APPEND mode
assert len(table.metadata.snapshots) == 1
assert table.metadata.snapshots[0].snapshot_id == table.metadata.current_snapshot_id
assert table.metadata.snapshots[0].parent_snapshot_id is None
assert table.metadata.snapshots[0].sequence_number == 1
assert table.metadata.snapshots[0].summary is not None
assert table.metadata.snapshots[0].summary.operation == Operation.APPEND
assert table.metadata.snapshots[0].summary['added-data-files'] == '1'
assert table.metadata.snapshots[0].summary['added-records'] == '1'
assert table.metadata.snapshots[0].summary['total-data-files'] == '1'
assert table.metadata.snapshots[0].summary['total-records'] == '1'

# read back the data
assert df == table.scan().to_arrow()


@pytest.mark.parametrize(
'catalog',
[
Expand Down
23 changes: 23 additions & 0 deletions tests/io/test_fsspec.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -15,10 +15,13 @@
# specific language governing permissions and limitations
# under the License.

import os
import tempfile
import uuid

import pytest
from botocore.awsrequest import AWSRequest
from fsspec.implementations.local import LocalFileSystem
from requests_mock import Mocker

from pyiceberg.exceptions import SignError
Expand All@@ -27,6 +30,26 @@
from pyiceberg.io.pyarrow import PyArrowFileIO


def test_fsspec_infer_local_fs_from_path(fsspec_fileio: FsspecFileIO) -> None:
"""Test path with `file` scheme and no scheme both use LocalFileSystem"""
assert isinstance(fsspec_fileio.new_output("file://tmp/warehouse")._fs, LocalFileSystem)
assert isinstance(fsspec_fileio.new_output("/tmp/warehouse")._fs, LocalFileSystem)


def test_fsspec_local_fs_can_create_path_without_parent_dir(fsspec_fileio: FsspecFileIO) -> None:
"""Test LocalFileSystem can create path without first creating the parent directories"""
with tempfile.TemporaryDirectory() as tmpdirname:
file_path = f"{tmpdirname}/foo/bar/baz.txt"
output_file = fsspec_fileio.new_output(file_path)
parent_path = os.path.dirname(file_path)
assert output_file._fs.exists(parent_path) is False
try:
with output_file.create() as f:
f.write(b"foo")
except Exception:
pytest.fail("Failed to write to file without parent directory")


@pytest.mark.s3
def test_fsspec_new_input_file(fsspec_fileio: FsspecFileIO) -> None:
"""Test creating a new input file from a fsspec file-io"""
Expand Down
20 changes: 20 additions & 0 deletions tests/io/test_pyarrow.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,26 @@
)


def test_pyarrow_infer_local_fs_from_path() -> None:
"""Test path with `file` scheme and no scheme both use LocalFileSystem"""
assert isinstance(PyArrowFileIO().new_output("file://tmp/warehouse")._filesystem, LocalFileSystem)
assert isinstance(PyArrowFileIO().new_output("/tmp/warehouse")._filesystem, LocalFileSystem)


def test_pyarrow_local_fs_can_create_path_without_parent_dir() -> None:
"""Test LocalFileSystem can create path without first creating the parent directories"""
with tempfile.TemporaryDirectory() as tmpdirname:
file_path = f"{tmpdirname}/foo/bar/baz.txt"
output_file = PyArrowFileIO().new_output(file_path)
parent_path = os.path.dirname(file_path)
assert output_file._filesystem.get_file_info(parent_path).type == FileType.NotFound
try:
with output_file.create() as f:
f.write(b"foo")
except Exception:
pytest.fail("Failed to write to file without parent directory")


def test_pyarrow_input_file() -> None:
"""Test reading a file using PyArrowFile"""

Expand Down
, '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
Fix writing to local filesystem#301

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

Fokko
merged 5 commits into

from

Jan 29, 2024
 Merged


Fix writing to local filesystem
#301
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
369df6e

add a test for writing data to disk


kevinjqliu Jan 24, 2024
5751b80

create dir first


kevinjqliu Jan 24, 2024
4101dd1

add fsspec test


kevinjqliu Jan 26, 2024
57d68f2

make lint


kevinjqliu Jan 26, 2024
242e365

add fs tests


kevinjqliu Jan 26, 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


2 changes: 1 addition & 1 deletion
pyiceberg/io/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -275,7 +275,7 @@ def delete(self, location: Union[str, InputFile, OutputFile]) -> None:
 "s3a": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "s3n": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "gs": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "hdfs": [ARROW_FILE_IO],
 "abfs": [FSSPEC_FILE_IO],
 "abfss": [FSSPEC_FILE_IO],
Expand Down

3 changes: 2 additions & 1 deletion
pyiceberg/io/fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -99,7 +99,7 @@ def s3v4_rest_signer(properties: Properties, request: AWSRequest, **_: Any) -> A


def _file(_: Properties) -> LocalFileSystem:
 return LocalFileSystem()
 return LocalFileSystem(auto_mkdir=True)


def _s3(properties: Properties) -> AbstractFileSystem:
Expand DownExpand Up@@ -173,6 +173,7 @@ def _adlfs(properties: Properties) -> AbstractFileSystem:


SCHEME_TO_FS = {
 "": _file,
@kevinjqliukevinjqliuJan 26, 2024

Copy link


Copy Markdown

ContributorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pyarrow defaults scheme to file when no scheme is present.
iceberg-python/pyiceberg/io/pyarrow.py

Line 322
in
8464d71
return"file", uri.netloc, os.path.abspath(location) 
we essentially do the same here
 "file": _file,
 "s3": _s3,
 "s3a": _s3,
Expand Down

11 changes: 8 additions & 3 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -172,6 +172,13 @@
T = TypeVar("T")


class PyArrowLocalFileSystem(pyarrow.fs.LocalFileSystem):
 def open_output_stream(self, path: str, *args: Any, **kwargs: Any) -> pyarrow.NativeFile:
 # In LocalFileSystem, parent directories must be first created before opening an output stream
 self.create_dir(os.path.dirname(path), recursive=True)
 return super().open_output_stream(path, *args, **kwargs)


class PyArrowFile(InputFile, OutputFile):
 """A combined InputFile and OutputFile implementation that uses a pyarrow filesystem to generate pyarrow.lib.NativeFile instances.

Expand DownExpand Up@@ -378,9 +385,7 @@ def _initialize_fs(self, scheme: str, netloc: Optional[str] = None) -> FileSyste

 return GcsFileSystem(**gcs_kwargs)
 elif scheme == "file":
 from pyarrow.fs import LocalFileSystem

 return LocalFileSystem()
 return PyArrowLocalFileSystem()
 else:
 raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")

Expand Down

62 changes: 60 additions & 2 deletions
tests/catalog/test_sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,6 +19,7 @@
from pathlib import Path
from typing import Generator, List

import pyarrow as pa
import pytest
from pytest import TempPathFactory
from pytest_lazyfixture import lazy_fixture
Expand All@@ -35,7 +36,10 @@
 NoSuchTableError,
 TableAlreadyExistsError,
)
from pyiceberg.io import FSSPEC_FILE_IO, PY_IO_IMPL
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.schema import Schema
from pyiceberg.table.snapshots import Operation
from pyiceberg.table.sorting import (
 NullOrder,
 SortDirection,
Expand DownExpand Up@@ -80,7 +84,7 @@ def catalog_memory(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -92,7 +96,7 @@ def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -102,6 +106,19 @@ def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, No
 catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite_fsspec(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 PY_IO_IMPL: FSSPEC_FILE_IO,
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
 catalog.create_tables()
 yield catalog
 catalog.destroy_tables()


def test_creation_with_no_uri() -> None:
 with pytest.raises(NoSuchPropertyException):
 SqlCatalog("test_ddb_catalog", not_uri="unused")
Expand DownExpand Up@@ -722,6 +739,47 @@ def test_commit_table(catalog: SqlCatalog, table_schema_nested: Schema, random_i
 assert new_schema.find_field("b").field_type == IntegerType()


@pytest.mark.parametrize(
 'catalog',
 [
 lazy_fixture('catalog_memory'),
 lazy_fixture('catalog_sqlite'),
 lazy_fixture('catalog_sqlite_without_rowcount'),
 lazy_fixture('catalog_sqlite_fsspec'),
 ],
)
def test_append_table(catalog: SqlCatalog, table_schema_simple: Schema, random_identifier: Identifier) -> None:
 database_name, _table_name = random_identifier
 catalog.create_namespace(database_name)
 table = catalog.create_table(random_identifier, table_schema_simple)

 df = pa.Table.from_pydict(
 {
 "foo": ["a"],
 "bar": [1],
 "baz": [True],
 },
 schema=schema_to_pyarrow(table_schema_simple),
 )

 table.append(df)

 # new snapshot is written in APPEND mode
 assert len(table.metadata.snapshots) == 1
 assert table.metadata.snapshots[0].snapshot_id == table.metadata.current_snapshot_id
 assert table.metadata.snapshots[0].parent_snapshot_id is None
 assert table.metadata.snapshots[0].sequence_number == 1
 assert table.metadata.snapshots[0].summary is not None
 assert table.metadata.snapshots[0].summary.operation == Operation.APPEND
 assert table.metadata.snapshots[0].summary['added-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['added-records'] == '1'
 assert table.metadata.snapshots[0].summary['total-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['total-records'] == '1'

 # read back the data
 assert df == table.scan().to_arrow()


@pytest.mark.parametrize(
 'catalog',
 [
Expand Down

23 changes: 23 additions & 0 deletions
tests/io/test_fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,10 +15,13 @@
# specific language governing permissions and limitations
# under the License.

import os
import tempfile
import uuid

import pytest
from botocore.awsrequest import AWSRequest
from fsspec.implementations.local import LocalFileSystem
from requests_mock import Mocker

from pyiceberg.exceptions import SignError
Expand All@@ -27,6 +30,26 @@
from pyiceberg.io.pyarrow import PyArrowFileIO


def test_fsspec_infer_local_fs_from_path(fsspec_fileio: FsspecFileIO) -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(fsspec_fileio.new_output("file://tmp/warehouse")._fs, LocalFileSystem)
 assert isinstance(fsspec_fileio.new_output("/tmp/warehouse")._fs, LocalFileSystem)


def test_fsspec_local_fs_can_create_path_without_parent_dir(fsspec_fileio: FsspecFileIO) -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = fsspec_fileio.new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._fs.exists(parent_path) is False
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


@pytest.mark.s3
def test_fsspec_new_input_file(fsspec_fileio: FsspecFileIO) -> None:
 """Test creating a new input file from a fsspec file-io"""
Expand Down

20 changes: 20 additions & 0 deletions
tests/io/test_pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,26 @@
)


def test_pyarrow_infer_local_fs_from_path() -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(PyArrowFileIO().new_output("file://tmp/warehouse")._filesystem, LocalFileSystem)
 assert isinstance(PyArrowFileIO().new_output("/tmp/warehouse")._filesystem, LocalFileSystem)


def test_pyarrow_local_fs_can_create_path_without_parent_dir() -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = PyArrowFileIO().new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._filesystem.get_file_info(parent_path).type == FileType.NotFound
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


def test_pyarrow_input_file() -> None:
 """Test reading a file using PyArrowFile"""

Expand Down

, '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
Fix writing to local filesystem#301

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

Fokko
merged 5 commits into

from

Jan 29, 2024
 Merged


Fix writing to local filesystem
#301
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
369df6e

add a test for writing data to disk


kevinjqliu Jan 24, 2024
5751b80

create dir first


kevinjqliu Jan 24, 2024
4101dd1

add fsspec test


kevinjqliu Jan 26, 2024
57d68f2

make lint


kevinjqliu Jan 26, 2024
242e365

add fs tests


kevinjqliu Jan 26, 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


2 changes: 1 addition & 1 deletion
pyiceberg/io/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -275,7 +275,7 @@ def delete(self, location: Union[str, InputFile, OutputFile]) -> None:
 "s3a": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "s3n": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "gs": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "hdfs": [ARROW_FILE_IO],
 "abfs": [FSSPEC_FILE_IO],
 "abfss": [FSSPEC_FILE_IO],
Expand Down

3 changes: 2 additions & 1 deletion
pyiceberg/io/fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -99,7 +99,7 @@ def s3v4_rest_signer(properties: Properties, request: AWSRequest, **_: Any) -> A


def _file(_: Properties) -> LocalFileSystem:
 return LocalFileSystem()
 return LocalFileSystem(auto_mkdir=True)


def _s3(properties: Properties) -> AbstractFileSystem:
Expand DownExpand Up@@ -173,6 +173,7 @@ def _adlfs(properties: Properties) -> AbstractFileSystem:


SCHEME_TO_FS = {
 "": _file,
@kevinjqliukevinjqliuJan 26, 2024

Copy link


Copy Markdown

ContributorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pyarrow defaults scheme to file when no scheme is present.
iceberg-python/pyiceberg/io/pyarrow.py

Line 322
in
8464d71
return"file", uri.netloc, os.path.abspath(location) 
we essentially do the same here
 "file": _file,
 "s3": _s3,
 "s3a": _s3,
Expand Down

11 changes: 8 additions & 3 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -172,6 +172,13 @@
T = TypeVar("T")


class PyArrowLocalFileSystem(pyarrow.fs.LocalFileSystem):
 def open_output_stream(self, path: str, *args: Any, **kwargs: Any) -> pyarrow.NativeFile:
 # In LocalFileSystem, parent directories must be first created before opening an output stream
 self.create_dir(os.path.dirname(path), recursive=True)
 return super().open_output_stream(path, *args, **kwargs)


class PyArrowFile(InputFile, OutputFile):
 """A combined InputFile and OutputFile implementation that uses a pyarrow filesystem to generate pyarrow.lib.NativeFile instances.

Expand DownExpand Up@@ -378,9 +385,7 @@ def _initialize_fs(self, scheme: str, netloc: Optional[str] = None) -> FileSyste

 return GcsFileSystem(**gcs_kwargs)
 elif scheme == "file":
 from pyarrow.fs import LocalFileSystem

 return LocalFileSystem()
 return PyArrowLocalFileSystem()
 else:
 raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")

Expand Down

62 changes: 60 additions & 2 deletions
tests/catalog/test_sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,6 +19,7 @@
from pathlib import Path
from typing import Generator, List

import pyarrow as pa
import pytest
from pytest import TempPathFactory
from pytest_lazyfixture import lazy_fixture
Expand All@@ -35,7 +36,10 @@
 NoSuchTableError,
 TableAlreadyExistsError,
)
from pyiceberg.io import FSSPEC_FILE_IO, PY_IO_IMPL
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.schema import Schema
from pyiceberg.table.snapshots import Operation
from pyiceberg.table.sorting import (
 NullOrder,
 SortDirection,
Expand DownExpand Up@@ -80,7 +84,7 @@ def catalog_memory(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -92,7 +96,7 @@ def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -102,6 +106,19 @@ def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, No
 catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite_fsspec(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 PY_IO_IMPL: FSSPEC_FILE_IO,
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
 catalog.create_tables()
 yield catalog
 catalog.destroy_tables()


def test_creation_with_no_uri() -> None:
 with pytest.raises(NoSuchPropertyException):
 SqlCatalog("test_ddb_catalog", not_uri="unused")
Expand DownExpand Up@@ -722,6 +739,47 @@ def test_commit_table(catalog: SqlCatalog, table_schema_nested: Schema, random_i
 assert new_schema.find_field("b").field_type == IntegerType()


@pytest.mark.parametrize(
 'catalog',
 [
 lazy_fixture('catalog_memory'),
 lazy_fixture('catalog_sqlite'),
 lazy_fixture('catalog_sqlite_without_rowcount'),
 lazy_fixture('catalog_sqlite_fsspec'),
 ],
)
def test_append_table(catalog: SqlCatalog, table_schema_simple: Schema, random_identifier: Identifier) -> None:
 database_name, _table_name = random_identifier
 catalog.create_namespace(database_name)
 table = catalog.create_table(random_identifier, table_schema_simple)

 df = pa.Table.from_pydict(
 {
 "foo": ["a"],
 "bar": [1],
 "baz": [True],
 },
 schema=schema_to_pyarrow(table_schema_simple),
 )

 table.append(df)

 # new snapshot is written in APPEND mode
 assert len(table.metadata.snapshots) == 1
 assert table.metadata.snapshots[0].snapshot_id == table.metadata.current_snapshot_id
 assert table.metadata.snapshots[0].parent_snapshot_id is None
 assert table.metadata.snapshots[0].sequence_number == 1
 assert table.metadata.snapshots[0].summary is not None
 assert table.metadata.snapshots[0].summary.operation == Operation.APPEND
 assert table.metadata.snapshots[0].summary['added-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['added-records'] == '1'
 assert table.metadata.snapshots[0].summary['total-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['total-records'] == '1'

 # read back the data
 assert df == table.scan().to_arrow()


@pytest.mark.parametrize(
 'catalog',
 [
Expand Down

23 changes: 23 additions & 0 deletions
tests/io/test_fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,10 +15,13 @@
# specific language governing permissions and limitations
# under the License.

import os
import tempfile
import uuid

import pytest
from botocore.awsrequest import AWSRequest
from fsspec.implementations.local import LocalFileSystem
from requests_mock import Mocker

from pyiceberg.exceptions import SignError
Expand All@@ -27,6 +30,26 @@
from pyiceberg.io.pyarrow import PyArrowFileIO


def test_fsspec_infer_local_fs_from_path(fsspec_fileio: FsspecFileIO) -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(fsspec_fileio.new_output("file://tmp/warehouse")._fs, LocalFileSystem)
 assert isinstance(fsspec_fileio.new_output("/tmp/warehouse")._fs, LocalFileSystem)


def test_fsspec_local_fs_can_create_path_without_parent_dir(fsspec_fileio: FsspecFileIO) -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = fsspec_fileio.new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._fs.exists(parent_path) is False
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


@pytest.mark.s3
def test_fsspec_new_input_file(fsspec_fileio: FsspecFileIO) -> None:
 """Test creating a new input file from a fsspec file-io"""
Expand Down

20 changes: 20 additions & 0 deletions
tests/io/test_pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,26 @@
)


def test_pyarrow_infer_local_fs_from_path() -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(PyArrowFileIO().new_output("file://tmp/warehouse")._filesystem, LocalFileSystem)
 assert isinstance(PyArrowFileIO().new_output("/tmp/warehouse")._filesystem, LocalFileSystem)


def test_pyarrow_local_fs_can_create_path_without_parent_dir() -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = PyArrowFileIO().new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._filesystem.get_file_info(parent_path).type == FileType.NotFound
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


def test_pyarrow_input_file() -> None:
 """Test reading a file using PyArrowFile"""

Expand Down

, '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
Fix writing to local filesystem#301

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

Fokko
merged 5 commits into

from

Jan 29, 2024
 Merged


Fix writing to local filesystem
#301
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
369df6e

add a test for writing data to disk


kevinjqliu Jan 24, 2024
5751b80

create dir first


kevinjqliu Jan 24, 2024
4101dd1

add fsspec test


kevinjqliu Jan 26, 2024
57d68f2

make lint


kevinjqliu Jan 26, 2024
242e365

add fs tests


kevinjqliu Jan 26, 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


2 changes: 1 addition & 1 deletion
pyiceberg/io/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -275,7 +275,7 @@ def delete(self, location: Union[str, InputFile, OutputFile]) -> None:
 "s3a": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "s3n": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "gs": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "hdfs": [ARROW_FILE_IO],
 "abfs": [FSSPEC_FILE_IO],
 "abfss": [FSSPEC_FILE_IO],
Expand Down

3 changes: 2 additions & 1 deletion
pyiceberg/io/fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -99,7 +99,7 @@ def s3v4_rest_signer(properties: Properties, request: AWSRequest, **_: Any) -> A


def _file(_: Properties) -> LocalFileSystem:
 return LocalFileSystem()
 return LocalFileSystem(auto_mkdir=True)


def _s3(properties: Properties) -> AbstractFileSystem:
Expand DownExpand Up@@ -173,6 +173,7 @@ def _adlfs(properties: Properties) -> AbstractFileSystem:


SCHEME_TO_FS = {
 "": _file,
@kevinjqliukevinjqliuJan 26, 2024

Copy link


Copy Markdown

ContributorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pyarrow defaults scheme to file when no scheme is present.
iceberg-python/pyiceberg/io/pyarrow.py

Line 322
in
8464d71
return"file", uri.netloc, os.path.abspath(location) 
we essentially do the same here
 "file": _file,
 "s3": _s3,
 "s3a": _s3,
Expand Down

11 changes: 8 additions & 3 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -172,6 +172,13 @@
T = TypeVar("T")


class PyArrowLocalFileSystem(pyarrow.fs.LocalFileSystem):
 def open_output_stream(self, path: str, *args: Any, **kwargs: Any) -> pyarrow.NativeFile:
 # In LocalFileSystem, parent directories must be first created before opening an output stream
 self.create_dir(os.path.dirname(path), recursive=True)
 return super().open_output_stream(path, *args, **kwargs)


class PyArrowFile(InputFile, OutputFile):
 """A combined InputFile and OutputFile implementation that uses a pyarrow filesystem to generate pyarrow.lib.NativeFile instances.

Expand DownExpand Up@@ -378,9 +385,7 @@ def _initialize_fs(self, scheme: str, netloc: Optional[str] = None) -> FileSyste

 return GcsFileSystem(**gcs_kwargs)
 elif scheme == "file":
 from pyarrow.fs import LocalFileSystem

 return LocalFileSystem()
 return PyArrowLocalFileSystem()
 else:
 raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")

Expand Down

62 changes: 60 additions & 2 deletions
tests/catalog/test_sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,6 +19,7 @@
from pathlib import Path
from typing import Generator, List

import pyarrow as pa
import pytest
from pytest import TempPathFactory
from pytest_lazyfixture import lazy_fixture
Expand All@@ -35,7 +36,10 @@
 NoSuchTableError,
 TableAlreadyExistsError,
)
from pyiceberg.io import FSSPEC_FILE_IO, PY_IO_IMPL
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.schema import Schema
from pyiceberg.table.snapshots import Operation
from pyiceberg.table.sorting import (
 NullOrder,
 SortDirection,
Expand DownExpand Up@@ -80,7 +84,7 @@ def catalog_memory(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -92,7 +96,7 @@ def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -102,6 +106,19 @@ def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, No
 catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite_fsspec(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 PY_IO_IMPL: FSSPEC_FILE_IO,
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
 catalog.create_tables()
 yield catalog
 catalog.destroy_tables()


def test_creation_with_no_uri() -> None:
 with pytest.raises(NoSuchPropertyException):
 SqlCatalog("test_ddb_catalog", not_uri="unused")
Expand DownExpand Up@@ -722,6 +739,47 @@ def test_commit_table(catalog: SqlCatalog, table_schema_nested: Schema, random_i
 assert new_schema.find_field("b").field_type == IntegerType()


@pytest.mark.parametrize(
 'catalog',
 [
 lazy_fixture('catalog_memory'),
 lazy_fixture('catalog_sqlite'),
 lazy_fixture('catalog_sqlite_without_rowcount'),
 lazy_fixture('catalog_sqlite_fsspec'),
 ],
)
def test_append_table(catalog: SqlCatalog, table_schema_simple: Schema, random_identifier: Identifier) -> None:
 database_name, _table_name = random_identifier
 catalog.create_namespace(database_name)
 table = catalog.create_table(random_identifier, table_schema_simple)

 df = pa.Table.from_pydict(
 {
 "foo": ["a"],
 "bar": [1],
 "baz": [True],
 },
 schema=schema_to_pyarrow(table_schema_simple),
 )

 table.append(df)

 # new snapshot is written in APPEND mode
 assert len(table.metadata.snapshots) == 1
 assert table.metadata.snapshots[0].snapshot_id == table.metadata.current_snapshot_id
 assert table.metadata.snapshots[0].parent_snapshot_id is None
 assert table.metadata.snapshots[0].sequence_number == 1
 assert table.metadata.snapshots[0].summary is not None
 assert table.metadata.snapshots[0].summary.operation == Operation.APPEND
 assert table.metadata.snapshots[0].summary['added-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['added-records'] == '1'
 assert table.metadata.snapshots[0].summary['total-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['total-records'] == '1'

 # read back the data
 assert df == table.scan().to_arrow()


@pytest.mark.parametrize(
 'catalog',
 [
Expand Down

23 changes: 23 additions & 0 deletions
tests/io/test_fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,10 +15,13 @@
# specific language governing permissions and limitations
# under the License.

import os
import tempfile
import uuid

import pytest
from botocore.awsrequest import AWSRequest
from fsspec.implementations.local import LocalFileSystem
from requests_mock import Mocker

from pyiceberg.exceptions import SignError
Expand All@@ -27,6 +30,26 @@
from pyiceberg.io.pyarrow import PyArrowFileIO


def test_fsspec_infer_local_fs_from_path(fsspec_fileio: FsspecFileIO) -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(fsspec_fileio.new_output("file://tmp/warehouse")._fs, LocalFileSystem)
 assert isinstance(fsspec_fileio.new_output("/tmp/warehouse")._fs, LocalFileSystem)


def test_fsspec_local_fs_can_create_path_without_parent_dir(fsspec_fileio: FsspecFileIO) -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = fsspec_fileio.new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._fs.exists(parent_path) is False
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


@pytest.mark.s3
def test_fsspec_new_input_file(fsspec_fileio: FsspecFileIO) -> None:
 """Test creating a new input file from a fsspec file-io"""
Expand Down

20 changes: 20 additions & 0 deletions
tests/io/test_pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,26 @@
)


def test_pyarrow_infer_local_fs_from_path() -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(PyArrowFileIO().new_output("file://tmp/warehouse")._filesystem, LocalFileSystem)
 assert isinstance(PyArrowFileIO().new_output("/tmp/warehouse")._filesystem, LocalFileSystem)


def test_pyarrow_local_fs_can_create_path_without_parent_dir() -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = PyArrowFileIO().new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._filesystem.get_file_info(parent_path).type == FileType.NotFound
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


def test_pyarrow_input_file() -> None:
 """Test reading a file using PyArrowFile"""

Expand Down

, '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
Fix writing to local filesystem#301

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

Fokko
merged 5 commits into

from

Jan 29, 2024
 Merged


Fix writing to local filesystem
#301
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
369df6e

add a test for writing data to disk


kevinjqliu Jan 24, 2024
5751b80

create dir first


kevinjqliu Jan 24, 2024
4101dd1

add fsspec test


kevinjqliu Jan 26, 2024
57d68f2

make lint


kevinjqliu Jan 26, 2024
242e365

add fs tests


kevinjqliu Jan 26, 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


2 changes: 1 addition & 1 deletion
pyiceberg/io/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -275,7 +275,7 @@ def delete(self, location: Union[str, InputFile, OutputFile]) -> None:
 "s3a": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "s3n": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "gs": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "hdfs": [ARROW_FILE_IO],
 "abfs": [FSSPEC_FILE_IO],
 "abfss": [FSSPEC_FILE_IO],
Expand Down

3 changes: 2 additions & 1 deletion
pyiceberg/io/fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -99,7 +99,7 @@ def s3v4_rest_signer(properties: Properties, request: AWSRequest, **_: Any) -> A


def _file(_: Properties) -> LocalFileSystem:
 return LocalFileSystem()
 return LocalFileSystem(auto_mkdir=True)


def _s3(properties: Properties) -> AbstractFileSystem:
Expand DownExpand Up@@ -173,6 +173,7 @@ def _adlfs(properties: Properties) -> AbstractFileSystem:


SCHEME_TO_FS = {
 "": _file,
@kevinjqliukevinjqliuJan 26, 2024

Copy link


Copy Markdown

ContributorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pyarrow defaults scheme to file when no scheme is present.
iceberg-python/pyiceberg/io/pyarrow.py

Line 322
in
8464d71
return"file", uri.netloc, os.path.abspath(location) 
we essentially do the same here
 "file": _file,
 "s3": _s3,
 "s3a": _s3,
Expand Down

11 changes: 8 additions & 3 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -172,6 +172,13 @@
T = TypeVar("T")


class PyArrowLocalFileSystem(pyarrow.fs.LocalFileSystem):
 def open_output_stream(self, path: str, *args: Any, **kwargs: Any) -> pyarrow.NativeFile:
 # In LocalFileSystem, parent directories must be first created before opening an output stream
 self.create_dir(os.path.dirname(path), recursive=True)
 return super().open_output_stream(path, *args, **kwargs)


class PyArrowFile(InputFile, OutputFile):
 """A combined InputFile and OutputFile implementation that uses a pyarrow filesystem to generate pyarrow.lib.NativeFile instances.

Expand DownExpand Up@@ -378,9 +385,7 @@ def _initialize_fs(self, scheme: str, netloc: Optional[str] = None) -> FileSyste

 return GcsFileSystem(**gcs_kwargs)
 elif scheme == "file":
 from pyarrow.fs import LocalFileSystem

 return LocalFileSystem()
 return PyArrowLocalFileSystem()
 else:
 raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")

Expand Down

62 changes: 60 additions & 2 deletions
tests/catalog/test_sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,6 +19,7 @@
from pathlib import Path
from typing import Generator, List

import pyarrow as pa
import pytest
from pytest import TempPathFactory
from pytest_lazyfixture import lazy_fixture
Expand All@@ -35,7 +36,10 @@
 NoSuchTableError,
 TableAlreadyExistsError,
)
from pyiceberg.io import FSSPEC_FILE_IO, PY_IO_IMPL
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.schema import Schema
from pyiceberg.table.snapshots import Operation
from pyiceberg.table.sorting import (
 NullOrder,
 SortDirection,
Expand DownExpand Up@@ -80,7 +84,7 @@ def catalog_memory(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -92,7 +96,7 @@ def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -102,6 +106,19 @@ def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, No
 catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite_fsspec(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 PY_IO_IMPL: FSSPEC_FILE_IO,
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
 catalog.create_tables()
 yield catalog
 catalog.destroy_tables()


def test_creation_with_no_uri() -> None:
 with pytest.raises(NoSuchPropertyException):
 SqlCatalog("test_ddb_catalog", not_uri="unused")
Expand DownExpand Up@@ -722,6 +739,47 @@ def test_commit_table(catalog: SqlCatalog, table_schema_nested: Schema, random_i
 assert new_schema.find_field("b").field_type == IntegerType()


@pytest.mark.parametrize(
 'catalog',
 [
 lazy_fixture('catalog_memory'),
 lazy_fixture('catalog_sqlite'),
 lazy_fixture('catalog_sqlite_without_rowcount'),
 lazy_fixture('catalog_sqlite_fsspec'),
 ],
)
def test_append_table(catalog: SqlCatalog, table_schema_simple: Schema, random_identifier: Identifier) -> None:
 database_name, _table_name = random_identifier
 catalog.create_namespace(database_name)
 table = catalog.create_table(random_identifier, table_schema_simple)

 df = pa.Table.from_pydict(
 {
 "foo": ["a"],
 "bar": [1],
 "baz": [True],
 },
 schema=schema_to_pyarrow(table_schema_simple),
 )

 table.append(df)

 # new snapshot is written in APPEND mode
 assert len(table.metadata.snapshots) == 1
 assert table.metadata.snapshots[0].snapshot_id == table.metadata.current_snapshot_id
 assert table.metadata.snapshots[0].parent_snapshot_id is None
 assert table.metadata.snapshots[0].sequence_number == 1
 assert table.metadata.snapshots[0].summary is not None
 assert table.metadata.snapshots[0].summary.operation == Operation.APPEND
 assert table.metadata.snapshots[0].summary['added-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['added-records'] == '1'
 assert table.metadata.snapshots[0].summary['total-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['total-records'] == '1'

 # read back the data
 assert df == table.scan().to_arrow()


@pytest.mark.parametrize(
 'catalog',
 [
Expand Down

23 changes: 23 additions & 0 deletions
tests/io/test_fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,10 +15,13 @@
# specific language governing permissions and limitations
# under the License.

import os
import tempfile
import uuid

import pytest
from botocore.awsrequest import AWSRequest
from fsspec.implementations.local import LocalFileSystem
from requests_mock import Mocker

from pyiceberg.exceptions import SignError
Expand All@@ -27,6 +30,26 @@
from pyiceberg.io.pyarrow import PyArrowFileIO


def test_fsspec_infer_local_fs_from_path(fsspec_fileio: FsspecFileIO) -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(fsspec_fileio.new_output("file://tmp/warehouse")._fs, LocalFileSystem)
 assert isinstance(fsspec_fileio.new_output("/tmp/warehouse")._fs, LocalFileSystem)


def test_fsspec_local_fs_can_create_path_without_parent_dir(fsspec_fileio: FsspecFileIO) -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = fsspec_fileio.new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._fs.exists(parent_path) is False
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


@pytest.mark.s3
def test_fsspec_new_input_file(fsspec_fileio: FsspecFileIO) -> None:
 """Test creating a new input file from a fsspec file-io"""
Expand Down

20 changes: 20 additions & 0 deletions
tests/io/test_pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,26 @@
)


def test_pyarrow_infer_local_fs_from_path() -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(PyArrowFileIO().new_output("file://tmp/warehouse")._filesystem, LocalFileSystem)
 assert isinstance(PyArrowFileIO().new_output("/tmp/warehouse")._filesystem, LocalFileSystem)


def test_pyarrow_local_fs_can_create_path_without_parent_dir() -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = PyArrowFileIO().new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._filesystem.get_file_info(parent_path).type == FileType.NotFound
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


def test_pyarrow_input_file() -> None:
 """Test reading a file using PyArrowFile"""

Expand Down

, '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
Fix writing to local filesystem#301

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

Fokko
merged 5 commits into

from

Jan 29, 2024
 Merged


Fix writing to local filesystem
#301
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
369df6e

add a test for writing data to disk


kevinjqliu Jan 24, 2024
5751b80

create dir first


kevinjqliu Jan 24, 2024
4101dd1

add fsspec test


kevinjqliu Jan 26, 2024
57d68f2

make lint


kevinjqliu Jan 26, 2024
242e365

add fs tests


kevinjqliu Jan 26, 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


2 changes: 1 addition & 1 deletion
pyiceberg/io/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -275,7 +275,7 @@ def delete(self, location: Union[str, InputFile, OutputFile]) -> None:
 "s3a": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "s3n": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "gs": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "hdfs": [ARROW_FILE_IO],
 "abfs": [FSSPEC_FILE_IO],
 "abfss": [FSSPEC_FILE_IO],
Expand Down

3 changes: 2 additions & 1 deletion
pyiceberg/io/fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -99,7 +99,7 @@ def s3v4_rest_signer(properties: Properties, request: AWSRequest, **_: Any) -> A


def _file(_: Properties) -> LocalFileSystem:
 return LocalFileSystem()
 return LocalFileSystem(auto_mkdir=True)


def _s3(properties: Properties) -> AbstractFileSystem:
Expand DownExpand Up@@ -173,6 +173,7 @@ def _adlfs(properties: Properties) -> AbstractFileSystem:


SCHEME_TO_FS = {
 "": _file,
@kevinjqliukevinjqliuJan 26, 2024

Copy link


Copy Markdown

ContributorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pyarrow defaults scheme to file when no scheme is present.
iceberg-python/pyiceberg/io/pyarrow.py

Line 322
in
8464d71
return"file", uri.netloc, os.path.abspath(location) 
we essentially do the same here
 "file": _file,
 "s3": _s3,
 "s3a": _s3,
Expand Down

11 changes: 8 additions & 3 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -172,6 +172,13 @@
T = TypeVar("T")


class PyArrowLocalFileSystem(pyarrow.fs.LocalFileSystem):
 def open_output_stream(self, path: str, *args: Any, **kwargs: Any) -> pyarrow.NativeFile:
 # In LocalFileSystem, parent directories must be first created before opening an output stream
 self.create_dir(os.path.dirname(path), recursive=True)
 return super().open_output_stream(path, *args, **kwargs)


class PyArrowFile(InputFile, OutputFile):
 """A combined InputFile and OutputFile implementation that uses a pyarrow filesystem to generate pyarrow.lib.NativeFile instances.

Expand DownExpand Up@@ -378,9 +385,7 @@ def _initialize_fs(self, scheme: str, netloc: Optional[str] = None) -> FileSyste

 return GcsFileSystem(**gcs_kwargs)
 elif scheme == "file":
 from pyarrow.fs import LocalFileSystem

 return LocalFileSystem()
 return PyArrowLocalFileSystem()
 else:
 raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")

Expand Down

62 changes: 60 additions & 2 deletions
tests/catalog/test_sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,6 +19,7 @@
from pathlib import Path
from typing import Generator, List

import pyarrow as pa
import pytest
from pytest import TempPathFactory
from pytest_lazyfixture import lazy_fixture
Expand All@@ -35,7 +36,10 @@
 NoSuchTableError,
 TableAlreadyExistsError,
)
from pyiceberg.io import FSSPEC_FILE_IO, PY_IO_IMPL
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.schema import Schema
from pyiceberg.table.snapshots import Operation
from pyiceberg.table.sorting import (
 NullOrder,
 SortDirection,
Expand DownExpand Up@@ -80,7 +84,7 @@ def catalog_memory(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -92,7 +96,7 @@ def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -102,6 +106,19 @@ def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, No
 catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite_fsspec(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 PY_IO_IMPL: FSSPEC_FILE_IO,
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
 catalog.create_tables()
 yield catalog
 catalog.destroy_tables()


def test_creation_with_no_uri() -> None:
 with pytest.raises(NoSuchPropertyException):
 SqlCatalog("test_ddb_catalog", not_uri="unused")
Expand DownExpand Up@@ -722,6 +739,47 @@ def test_commit_table(catalog: SqlCatalog, table_schema_nested: Schema, random_i
 assert new_schema.find_field("b").field_type == IntegerType()


@pytest.mark.parametrize(
 'catalog',
 [
 lazy_fixture('catalog_memory'),
 lazy_fixture('catalog_sqlite'),
 lazy_fixture('catalog_sqlite_without_rowcount'),
 lazy_fixture('catalog_sqlite_fsspec'),
 ],
)
def test_append_table(catalog: SqlCatalog, table_schema_simple: Schema, random_identifier: Identifier) -> None:
 database_name, _table_name = random_identifier
 catalog.create_namespace(database_name)
 table = catalog.create_table(random_identifier, table_schema_simple)

 df = pa.Table.from_pydict(
 {
 "foo": ["a"],
 "bar": [1],
 "baz": [True],
 },
 schema=schema_to_pyarrow(table_schema_simple),
 )

 table.append(df)

 # new snapshot is written in APPEND mode
 assert len(table.metadata.snapshots) == 1
 assert table.metadata.snapshots[0].snapshot_id == table.metadata.current_snapshot_id
 assert table.metadata.snapshots[0].parent_snapshot_id is None
 assert table.metadata.snapshots[0].sequence_number == 1
 assert table.metadata.snapshots[0].summary is not None
 assert table.metadata.snapshots[0].summary.operation == Operation.APPEND
 assert table.metadata.snapshots[0].summary['added-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['added-records'] == '1'
 assert table.metadata.snapshots[0].summary['total-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['total-records'] == '1'

 # read back the data
 assert df == table.scan().to_arrow()


@pytest.mark.parametrize(
 'catalog',
 [
Expand Down

23 changes: 23 additions & 0 deletions
tests/io/test_fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,10 +15,13 @@
# specific language governing permissions and limitations
# under the License.

import os
import tempfile
import uuid

import pytest
from botocore.awsrequest import AWSRequest
from fsspec.implementations.local import LocalFileSystem
from requests_mock import Mocker

from pyiceberg.exceptions import SignError
Expand All@@ -27,6 +30,26 @@
from pyiceberg.io.pyarrow import PyArrowFileIO


def test_fsspec_infer_local_fs_from_path(fsspec_fileio: FsspecFileIO) -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(fsspec_fileio.new_output("file://tmp/warehouse")._fs, LocalFileSystem)
 assert isinstance(fsspec_fileio.new_output("/tmp/warehouse")._fs, LocalFileSystem)


def test_fsspec_local_fs_can_create_path_without_parent_dir(fsspec_fileio: FsspecFileIO) -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = fsspec_fileio.new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._fs.exists(parent_path) is False
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


@pytest.mark.s3
def test_fsspec_new_input_file(fsspec_fileio: FsspecFileIO) -> None:
 """Test creating a new input file from a fsspec file-io"""
Expand Down

20 changes: 20 additions & 0 deletions
tests/io/test_pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,26 @@
)


def test_pyarrow_infer_local_fs_from_path() -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(PyArrowFileIO().new_output("file://tmp/warehouse")._filesystem, LocalFileSystem)
 assert isinstance(PyArrowFileIO().new_output("/tmp/warehouse")._filesystem, LocalFileSystem)


def test_pyarrow_local_fs_can_create_path_without_parent_dir() -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = PyArrowFileIO().new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._filesystem.get_file_info(parent_path).type == FileType.NotFound
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


def test_pyarrow_input_file() -> None:
 """Test reading a file using PyArrowFile"""

Expand Down

, '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
Fix writing to local filesystem#301

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

Fokko
merged 5 commits into

from

Jan 29, 2024
 Merged


Fix writing to local filesystem
#301
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
369df6e

add a test for writing data to disk


kevinjqliu Jan 24, 2024
5751b80

create dir first


kevinjqliu Jan 24, 2024
4101dd1

add fsspec test


kevinjqliu Jan 26, 2024
57d68f2

make lint


kevinjqliu Jan 26, 2024
242e365

add fs tests


kevinjqliu Jan 26, 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


2 changes: 1 addition & 1 deletion
pyiceberg/io/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -275,7 +275,7 @@ def delete(self, location: Union[str, InputFile, OutputFile]) -> None:
 "s3a": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "s3n": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "gs": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "hdfs": [ARROW_FILE_IO],
 "abfs": [FSSPEC_FILE_IO],
 "abfss": [FSSPEC_FILE_IO],
Expand Down

3 changes: 2 additions & 1 deletion
pyiceberg/io/fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -99,7 +99,7 @@ def s3v4_rest_signer(properties: Properties, request: AWSRequest, **_: Any) -> A


def _file(_: Properties) -> LocalFileSystem:
 return LocalFileSystem()
 return LocalFileSystem(auto_mkdir=True)


def _s3(properties: Properties) -> AbstractFileSystem:
Expand DownExpand Up@@ -173,6 +173,7 @@ def _adlfs(properties: Properties) -> AbstractFileSystem:


SCHEME_TO_FS = {
 "": _file,
@kevinjqliukevinjqliuJan 26, 2024

Copy link


Copy Markdown

ContributorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pyarrow defaults scheme to file when no scheme is present.
iceberg-python/pyiceberg/io/pyarrow.py

Line 322
in
8464d71
return"file", uri.netloc, os.path.abspath(location) 
we essentially do the same here
 "file": _file,
 "s3": _s3,
 "s3a": _s3,
Expand Down

11 changes: 8 additions & 3 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -172,6 +172,13 @@
T = TypeVar("T")


class PyArrowLocalFileSystem(pyarrow.fs.LocalFileSystem):
 def open_output_stream(self, path: str, *args: Any, **kwargs: Any) -> pyarrow.NativeFile:
 # In LocalFileSystem, parent directories must be first created before opening an output stream
 self.create_dir(os.path.dirname(path), recursive=True)
 return super().open_output_stream(path, *args, **kwargs)


class PyArrowFile(InputFile, OutputFile):
 """A combined InputFile and OutputFile implementation that uses a pyarrow filesystem to generate pyarrow.lib.NativeFile instances.

Expand DownExpand Up@@ -378,9 +385,7 @@ def _initialize_fs(self, scheme: str, netloc: Optional[str] = None) -> FileSyste

 return GcsFileSystem(**gcs_kwargs)
 elif scheme == "file":
 from pyarrow.fs import LocalFileSystem

 return LocalFileSystem()
 return PyArrowLocalFileSystem()
 else:
 raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")

Expand Down

62 changes: 60 additions & 2 deletions
tests/catalog/test_sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,6 +19,7 @@
from pathlib import Path
from typing import Generator, List

import pyarrow as pa
import pytest
from pytest import TempPathFactory
from pytest_lazyfixture import lazy_fixture
Expand All@@ -35,7 +36,10 @@
 NoSuchTableError,
 TableAlreadyExistsError,
)
from pyiceberg.io import FSSPEC_FILE_IO, PY_IO_IMPL
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.schema import Schema
from pyiceberg.table.snapshots import Operation
from pyiceberg.table.sorting import (
 NullOrder,
 SortDirection,
Expand DownExpand Up@@ -80,7 +84,7 @@ def catalog_memory(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -92,7 +96,7 @@ def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -102,6 +106,19 @@ def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, No
 catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite_fsspec(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 PY_IO_IMPL: FSSPEC_FILE_IO,
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
 catalog.create_tables()
 yield catalog
 catalog.destroy_tables()


def test_creation_with_no_uri() -> None:
 with pytest.raises(NoSuchPropertyException):
 SqlCatalog("test_ddb_catalog", not_uri="unused")
Expand DownExpand Up@@ -722,6 +739,47 @@ def test_commit_table(catalog: SqlCatalog, table_schema_nested: Schema, random_i
 assert new_schema.find_field("b").field_type == IntegerType()


@pytest.mark.parametrize(
 'catalog',
 [
 lazy_fixture('catalog_memory'),
 lazy_fixture('catalog_sqlite'),
 lazy_fixture('catalog_sqlite_without_rowcount'),
 lazy_fixture('catalog_sqlite_fsspec'),
 ],
)
def test_append_table(catalog: SqlCatalog, table_schema_simple: Schema, random_identifier: Identifier) -> None:
 database_name, _table_name = random_identifier
 catalog.create_namespace(database_name)
 table = catalog.create_table(random_identifier, table_schema_simple)

 df = pa.Table.from_pydict(
 {
 "foo": ["a"],
 "bar": [1],
 "baz": [True],
 },
 schema=schema_to_pyarrow(table_schema_simple),
 )

 table.append(df)

 # new snapshot is written in APPEND mode
 assert len(table.metadata.snapshots) == 1
 assert table.metadata.snapshots[0].snapshot_id == table.metadata.current_snapshot_id
 assert table.metadata.snapshots[0].parent_snapshot_id is None
 assert table.metadata.snapshots[0].sequence_number == 1
 assert table.metadata.snapshots[0].summary is not None
 assert table.metadata.snapshots[0].summary.operation == Operation.APPEND
 assert table.metadata.snapshots[0].summary['added-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['added-records'] == '1'
 assert table.metadata.snapshots[0].summary['total-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['total-records'] == '1'

 # read back the data
 assert df == table.scan().to_arrow()


@pytest.mark.parametrize(
 'catalog',
 [
Expand Down

23 changes: 23 additions & 0 deletions
tests/io/test_fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,10 +15,13 @@
# specific language governing permissions and limitations
# under the License.

import os
import tempfile
import uuid

import pytest
from botocore.awsrequest import AWSRequest
from fsspec.implementations.local import LocalFileSystem
from requests_mock import Mocker

from pyiceberg.exceptions import SignError
Expand All@@ -27,6 +30,26 @@
from pyiceberg.io.pyarrow import PyArrowFileIO


def test_fsspec_infer_local_fs_from_path(fsspec_fileio: FsspecFileIO) -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(fsspec_fileio.new_output("file://tmp/warehouse")._fs, LocalFileSystem)
 assert isinstance(fsspec_fileio.new_output("/tmp/warehouse")._fs, LocalFileSystem)


def test_fsspec_local_fs_can_create_path_without_parent_dir(fsspec_fileio: FsspecFileIO) -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = fsspec_fileio.new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._fs.exists(parent_path) is False
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


@pytest.mark.s3
def test_fsspec_new_input_file(fsspec_fileio: FsspecFileIO) -> None:
 """Test creating a new input file from a fsspec file-io"""
Expand Down

20 changes: 20 additions & 0 deletions
tests/io/test_pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,26 @@
)


def test_pyarrow_infer_local_fs_from_path() -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(PyArrowFileIO().new_output("file://tmp/warehouse")._filesystem, LocalFileSystem)
 assert isinstance(PyArrowFileIO().new_output("/tmp/warehouse")._filesystem, LocalFileSystem)


def test_pyarrow_local_fs_can_create_path_without_parent_dir() -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = PyArrowFileIO().new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._filesystem.get_file_info(parent_path).type == FileType.NotFound
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


def test_pyarrow_input_file() -> None:
 """Test reading a file using PyArrowFile"""

Expand Down

, '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
Fix writing to local filesystem#301

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

Fokko
merged 5 commits into

from

Jan 29, 2024
 Merged


Fix writing to local filesystem
#301
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


5 commits


Select commit
Hold shift + click to select a range
369df6e

add a test for writing data to disk


kevinjqliu Jan 24, 2024
5751b80

create dir first


kevinjqliu Jan 24, 2024
4101dd1

add fsspec test


kevinjqliu Jan 26, 2024
57d68f2

make lint


kevinjqliu Jan 26, 2024
242e365

add fs tests


kevinjqliu Jan 26, 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


2 changes: 1 addition & 1 deletion
pyiceberg/io/__init__.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -275,7 +275,7 @@ def delete(self, location: Union[str, InputFile, OutputFile]) -> None:
 "s3a": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "s3n": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "gs": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO],
 "file": [ARROW_FILE_IO, FSSPEC_FILE_IO],
 "hdfs": [ARROW_FILE_IO],
 "abfs": [FSSPEC_FILE_IO],
 "abfss": [FSSPEC_FILE_IO],
Expand Down

3 changes: 2 additions & 1 deletion
pyiceberg/io/fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -99,7 +99,7 @@ def s3v4_rest_signer(properties: Properties, request: AWSRequest, **_: Any) -> A


def _file(_: Properties) -> LocalFileSystem:
 return LocalFileSystem()
 return LocalFileSystem(auto_mkdir=True)


def _s3(properties: Properties) -> AbstractFileSystem:
Expand DownExpand Up@@ -173,6 +173,7 @@ def _adlfs(properties: Properties) -> AbstractFileSystem:


SCHEME_TO_FS = {
 "": _file,
@kevinjqliukevinjqliuJan 26, 2024

Copy link


Copy Markdown

ContributorAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pyarrow defaults scheme to file when no scheme is present.
iceberg-python/pyiceberg/io/pyarrow.py

Line 322
in
8464d71
return"file", uri.netloc, os.path.abspath(location) 
we essentially do the same here
 "file": _file,
 "s3": _s3,
 "s3a": _s3,
Expand Down

11 changes: 8 additions & 3 deletions
pyiceberg/io/pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -172,6 +172,13 @@
T = TypeVar("T")


class PyArrowLocalFileSystem(pyarrow.fs.LocalFileSystem):
 def open_output_stream(self, path: str, *args: Any, **kwargs: Any) -> pyarrow.NativeFile:
 # In LocalFileSystem, parent directories must be first created before opening an output stream
 self.create_dir(os.path.dirname(path), recursive=True)
 return super().open_output_stream(path, *args, **kwargs)


class PyArrowFile(InputFile, OutputFile):
 """A combined InputFile and OutputFile implementation that uses a pyarrow filesystem to generate pyarrow.lib.NativeFile instances.

Expand DownExpand Up@@ -378,9 +385,7 @@ def _initialize_fs(self, scheme: str, netloc: Optional[str] = None) -> FileSyste

 return GcsFileSystem(**gcs_kwargs)
 elif scheme == "file":
 from pyarrow.fs import LocalFileSystem

 return LocalFileSystem()
 return PyArrowLocalFileSystem()
 else:
 raise ValueError(f"Unrecognized filesystem type in URI: {scheme}")

Expand Down

62 changes: 60 additions & 2 deletions
tests/catalog/test_sql.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -19,6 +19,7 @@
from pathlib import Path
from typing import Generator, List

import pyarrow as pa
import pytest
from pytest import TempPathFactory
from pytest_lazyfixture import lazy_fixture
Expand All@@ -35,7 +36,10 @@
 NoSuchTableError,
 TableAlreadyExistsError,
)
from pyiceberg.io import FSSPEC_FILE_IO, PY_IO_IMPL
from pyiceberg.io.pyarrow import schema_to_pyarrow
from pyiceberg.schema import Schema
from pyiceberg.table.snapshots import Operation
from pyiceberg.table.sorting import (
 NullOrder,
 SortDirection,
Expand DownExpand Up@@ -80,7 +84,7 @@ def catalog_memory(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -92,7 +96,7 @@ def catalog_sqlite(warehouse: Path) -> Generator[SqlCatalog, None, None]:
@pytest.fixture(scope="module")
def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": "sqlite:////tmp/sql-catalog.db",
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
Expand All@@ -102,6 +106,19 @@ def catalog_sqlite_without_rowcount(warehouse: Path) -> Generator[SqlCatalog, No
 catalog.destroy_tables()


@pytest.fixture(scope="module")
def catalog_sqlite_fsspec(warehouse: Path) -> Generator[SqlCatalog, None, None]:
 props = {
 "uri": f"sqlite:////{warehouse}/sql-catalog.db",
 "warehouse": f"file://{warehouse}",
 PY_IO_IMPL: FSSPEC_FILE_IO,
 }
 catalog = SqlCatalog("test_sql_catalog", **props)
 catalog.create_tables()
 yield catalog
 catalog.destroy_tables()


def test_creation_with_no_uri() -> None:
 with pytest.raises(NoSuchPropertyException):
 SqlCatalog("test_ddb_catalog", not_uri="unused")
Expand DownExpand Up@@ -722,6 +739,47 @@ def test_commit_table(catalog: SqlCatalog, table_schema_nested: Schema, random_i
 assert new_schema.find_field("b").field_type == IntegerType()


@pytest.mark.parametrize(
 'catalog',
 [
 lazy_fixture('catalog_memory'),
 lazy_fixture('catalog_sqlite'),
 lazy_fixture('catalog_sqlite_without_rowcount'),
 lazy_fixture('catalog_sqlite_fsspec'),
 ],
)
def test_append_table(catalog: SqlCatalog, table_schema_simple: Schema, random_identifier: Identifier) -> None:
 database_name, _table_name = random_identifier
 catalog.create_namespace(database_name)
 table = catalog.create_table(random_identifier, table_schema_simple)

 df = pa.Table.from_pydict(
 {
 "foo": ["a"],
 "bar": [1],
 "baz": [True],
 },
 schema=schema_to_pyarrow(table_schema_simple),
 )

 table.append(df)

 # new snapshot is written in APPEND mode
 assert len(table.metadata.snapshots) == 1
 assert table.metadata.snapshots[0].snapshot_id == table.metadata.current_snapshot_id
 assert table.metadata.snapshots[0].parent_snapshot_id is None
 assert table.metadata.snapshots[0].sequence_number == 1
 assert table.metadata.snapshots[0].summary is not None
 assert table.metadata.snapshots[0].summary.operation == Operation.APPEND
 assert table.metadata.snapshots[0].summary['added-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['added-records'] == '1'
 assert table.metadata.snapshots[0].summary['total-data-files'] == '1'
 assert table.metadata.snapshots[0].summary['total-records'] == '1'

 # read back the data
 assert df == table.scan().to_arrow()


@pytest.mark.parametrize(
 'catalog',
 [
Expand Down

23 changes: 23 additions & 0 deletions
tests/io/test_fsspec.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -15,10 +15,13 @@
# specific language governing permissions and limitations
# under the License.

import os
import tempfile
import uuid

import pytest
from botocore.awsrequest import AWSRequest
from fsspec.implementations.local import LocalFileSystem
from requests_mock import Mocker

from pyiceberg.exceptions import SignError
Expand All@@ -27,6 +30,26 @@
from pyiceberg.io.pyarrow import PyArrowFileIO


def test_fsspec_infer_local_fs_from_path(fsspec_fileio: FsspecFileIO) -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(fsspec_fileio.new_output("file://tmp/warehouse")._fs, LocalFileSystem)
 assert isinstance(fsspec_fileio.new_output("/tmp/warehouse")._fs, LocalFileSystem)


def test_fsspec_local_fs_can_create_path_without_parent_dir(fsspec_fileio: FsspecFileIO) -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = fsspec_fileio.new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._fs.exists(parent_path) is False
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


@pytest.mark.s3
def test_fsspec_new_input_file(fsspec_fileio: FsspecFileIO) -> None:
 """Test creating a new input file from a fsspec file-io"""
Expand Down

20 changes: 20 additions & 0 deletions
tests/io/test_pyarrow.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,26 @@
)


def test_pyarrow_infer_local_fs_from_path() -> None:
 """Test path with `file` scheme and no scheme both use LocalFileSystem"""
 assert isinstance(PyArrowFileIO().new_output("file://tmp/warehouse")._filesystem, LocalFileSystem)
 assert isinstance(PyArrowFileIO().new_output("/tmp/warehouse")._filesystem, LocalFileSystem)


def test_pyarrow_local_fs_can_create_path_without_parent_dir() -> None:
 """Test LocalFileSystem can create path without first creating the parent directories"""
 with tempfile.TemporaryDirectory() as tmpdirname:
 file_path = f"{tmpdirname}/foo/bar/baz.txt"
 output_file = PyArrowFileIO().new_output(file_path)
 parent_path = os.path.dirname(file_path)
 assert output_file._filesystem.get_file_info(parent_path).type == FileType.NotFound
 try:
 with output_file.create() as f:
 f.write(b"foo")
 except Exception:
 pytest.fail("Failed to write to file without parent directory")


def test_pyarrow_input_file() -> None:
 """Test reading a file using PyArrowFile"""

Expand Down