Merge pull request #66 from mabel-dev/0.5.6

plan/exec sep, phase 1

Author: joocer

dev/build_counter.py

@@ -29,7 +29,7 @@ class VersionStatus(Enum):
 
 __major_version__ = 0
 __minor_version__ = 5
-__revision_version__ = 8
+__revision_version__ = 9
 __author__ = "@joocer"
 __status__ = VersionStatus.RELEASE
 

opteryx/__version__.py

@@ -1,11 +1,11 @@
 # THIS FILE IS AUTOMATICALLY UPDATED DURING THE BUILD PROCESS
 # DO NOT EDIT THIS FILE DIRECTLY
 
-__build__ = 137
+__build__ = 139
 __author__ = "@joocer"
-__version__ = "0.5.8"
+__version__ = "0.5.9"
 __lib__ = "opteryx-core"
-__build_date__ = "2026-01-03T20:41:15.799632+00:00Z"
+__build_date__ = "2026-01-04T16:13:49.116825+00:00Z"
 
 # Store the version here so:
 # 1) we don't load dependencies by storing it in __init__.py

opteryx/connectors/filesystem_connector.py

@@ -11,7 +11,6 @@
 from concurrent.futures import wait
 from threading import Lock
 from typing import Dict
-from typing import List
 from typing import Optional
 from typing import Tuple
 
@@ -108,6 +107,26 @@ def get_executor(self):
             )
         return FileSystemTable._executor
 
+    def get_list_of_blob_names(self, prefix: str, predicates=None):
+        """
+        Get list of blob names (file paths) matching the prefix.
+
+        Args:
+            prefix: Directory/path prefix to list files from
+            predicates: Optional predicates (not used for file listing)
+
+        Returns:
+            List of file paths
+        """
+        from pyarrow.fs import FileSelector
+
+        # Create file selector to list files recursively
+        selector = FileSelector(prefix, recursive=True)
+        file_infos = self.filesystem.get_file_info(selector)
+
+        # Extract paths from FileInfo objects
+        return [info.path for info in file_infos]
+
     def read_blob(
         self, *, blob_name: str, decoder, just_schema=False, projection=None, selection=None
     ):
@@ -185,32 +204,6 @@ def blocking_read():
         telemetry.bytes_read += len(data)
         return ref
 
-    def get_list_of_blob_names(self, *, prefix: str, predicates: list = []) -> List[str]:
-        """
-        List all blobs matching the prefix.
-
-        Args:
-            prefix: Path prefix to search
-            predicates: Optional predicates for filtering (subclasses may use this)
-
-        Returns:
-            List of blob paths
-        """
-        from pyarrow.fs import FileSelector
-
-        # Use filesystem's file listing
-        selector = FileSelector(prefix, recursive=True)
-        file_infos = self.filesystem.get_file_info(selector)
-
-        # Filter for valid file extensions
-        blob_names = [
-            info.path
-            for info in file_infos
-            if info.is_file and info.path.endswith(TUPLE_OF_VALID_EXTENSIONS)
-        ]
-
-        return blob_names
-
     def read_dataset(
         self,
         columns: list = None,

opteryx/connectors/io_systems/__init__.py

@@ -13,4 +13,44 @@
     "OpteryxLocalFileSystem",
     "OpteryxGcsFileSystem",
     "OpteryxS3FileSystem",
+    "create_filesystem",
 ]
+
+
+def create_filesystem(protocol: str):
+    """
+    Factory function to instantiate appropriate filesystem based on protocol.
+
+    Used by execution operators to create filesystem from file path protocol prefix.
+    This enables generic execution that works across all storage types.
+
+    Args:
+        protocol: Protocol string from file path (e.g., "gs", "s3", "file")
+
+    Returns:
+        Appropriate filesystem instance
+
+    Raises:
+        ValueError: If protocol is not supported
+
+    Example:
+        >>> protocol = "gs"  # from "gs://bucket/file.parquet"
+        >>> fs = create_filesystem(protocol)
+        >>> # fs is an OpteryxGcsFileSystem instance
+    """
+    protocol_map = {
+        "gs": OpteryxGcsFileSystem,
+        "gcs": OpteryxGcsFileSystem,
+        "s3": OpteryxS3FileSystem,
+        "file": OpteryxLocalFileSystem,
+        "": OpteryxLocalFileSystem,  # No protocol = local file
+    }
+
+    if protocol not in protocol_map:
+        raise ValueError(
+            f"Unsupported storage protocol: {protocol}. "
+            f"Supported protocols: {list(protocol_map.keys())}"
+        )
+
+    filesystem_class = protocol_map[protocol]
+    return filesystem_class()

opteryx/connectors/io_systems/gcs_filesystem.py

@@ -148,11 +148,9 @@ def open_input_file(self, path: str):
     async def async_read_blob(self, *, blob_name, pool, session, statistics, **kwargs):
         import asyncio
 
-        from opteryx import system_statistics
+        from opteryx import system_telemetry as system_statistics
         from opteryx.utils import paths
 
-        print("async read blob:", blob_name)
-
         # strip gs:// prefix
         if blob_name.startswith("gs://"):
             blob_name = blob_name[5:]