Add S3FS Cursor for lightweight CSV-based result reading

Implements Issue #272: Add a new cursor type that reads CSV results from S3
using Python's standard csv module and PyAthena's S3FileSystem, without
requiring pandas or pyarrow dependencies.

New features:
- S3FSCursor: Synchronous cursor for reading CSV/TXT results from S3
- AsyncS3FSCursor: Asynchronous cursor using concurrent.futures
- AthenaS3FSResultSet: Streaming CSV reader with type conversion
- DefaultS3FSTypeConverter: Type converter for CSV-based results
- SQLAlchemy dialect: awsathena+s3fs:// connection URL support

Also adds rowcount property to WithResultSet mixin for CTAS support,
benefiting all cursor types (base, pandas, arrow, s3fs).

Closes #272

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: laughingman7743

pyathena/result_set.py

@@ -736,3 +736,16 @@ def result_reuse_minutes(self) -> Optional[int]:
         if not self.result_set:
             return None
         return self.result_set.result_reuse_minutes
+
+    @property
+    def rowcount(self) -> int:
+        """Get the number of rows affected by the last operation.
+
+        For SELECT statements, this returns -1 as per DB API 2.0 specification.
+        For DML operations (INSERT, UPDATE, DELETE) and CTAS, this returns
+        the number of affected rows.
+
+        Returns:
+            The number of rows, or -1 if not applicable or unknown.
+        """
+        return self.result_set.rowcount if self.result_set else -1

pyathena/s3fs/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

pyathena/s3fs/async_cursor.py

@@ -0,0 +1,218 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import logging
+from concurrent.futures import Future
+from multiprocessing import cpu_count
+from typing import Any, Dict, List, Optional, Tuple, Union, cast
+
+from pyathena.async_cursor import AsyncCursor
+from pyathena.common import CursorIterator
+from pyathena.error import ProgrammingError
+from pyathena.model import AthenaQueryExecution
+from pyathena.s3fs.converter import DefaultS3FSTypeConverter
+from pyathena.s3fs.result_set import AthenaS3FSResultSet
+
+_logger = logging.getLogger(__name__)
+
+
+class AsyncS3FSCursor(AsyncCursor):
+    """Asynchronous cursor that reads CSV results via S3FileSystem.
+
+    This cursor extends AsyncCursor to provide asynchronous query execution
+    with results read via Python's standard csv module and PyAthena's S3FileSystem.
+    It's a lightweight alternative when pandas/pyarrow are not needed.
+
+    Features:
+        - Asynchronous query execution with concurrent futures
+        - Uses Python's standard csv module for parsing
+        - Uses PyAthena's S3FileSystem for S3 access
+        - No external dependencies beyond boto3
+        - Memory-efficient streaming for large datasets
+
+    Attributes:
+        arraysize: Number of rows to fetch per batch (configurable).
+
+    Example:
+        >>> from pyathena.s3fs.async_cursor import AsyncS3FSCursor
+        >>>
+        >>> cursor = connection.cursor(AsyncS3FSCursor)
+        >>> query_id, future = cursor.execute("SELECT * FROM my_table")
+        >>>
+        >>> # Get result when ready
+        >>> result_set = future.result()
+        >>> rows = result_set.fetchall()
+
+    Note:
+        This cursor does not require pandas or pyarrow.
+    """
+
+    def __init__(
+        self,
+        s3_staging_dir: Optional[str] = None,
+        schema_name: Optional[str] = None,
+        catalog_name: Optional[str] = None,
+        work_group: Optional[str] = None,
+        poll_interval: float = 1,
+        encryption_option: Optional[str] = None,
+        kms_key: Optional[str] = None,
+        kill_on_interrupt: bool = True,
+        max_workers: int = (cpu_count() or 1) * 5,
+        arraysize: int = CursorIterator.DEFAULT_FETCH_SIZE,
+        result_reuse_enable: bool = False,
+        result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
+        **kwargs,
+    ) -> None:
+        """Initialize an AsyncS3FSCursor.
+
+        Args:
+            s3_staging_dir: S3 location for query results.
+            schema_name: Default schema name.
+            catalog_name: Default catalog name.
+            work_group: Athena workgroup name.
+            poll_interval: Query status polling interval in seconds.
+            encryption_option: S3 encryption option (SSE_S3, SSE_KMS, CSE_KMS).
+            kms_key: KMS key ARN for encryption.
+            kill_on_interrupt: Cancel running query on keyboard interrupt.
+            max_workers: Maximum number of workers for concurrent execution.
+            arraysize: Number of rows to fetch per batch.
+            result_reuse_enable: Enable Athena query result reuse.
+            result_reuse_minutes: Minutes to reuse cached results.
+            **kwargs: Additional connection parameters.
+
+        Example:
+            >>> cursor = connection.cursor(AsyncS3FSCursor)
+            >>> query_id, future = cursor.execute("SELECT * FROM my_table")
+        """
+        super().__init__(
+            s3_staging_dir=s3_staging_dir,
+            schema_name=schema_name,
+            catalog_name=catalog_name,
+            work_group=work_group,
+            poll_interval=poll_interval,
+            encryption_option=encryption_option,
+            kms_key=kms_key,
+            kill_on_interrupt=kill_on_interrupt,
+            max_workers=max_workers,
+            arraysize=arraysize,
+            result_reuse_enable=result_reuse_enable,
+            result_reuse_minutes=result_reuse_minutes,
+            **kwargs,
+        )
+
+    @staticmethod
+    def get_default_converter(
+        unload: bool = False,  # noqa: ARG004
+    ) -> DefaultS3FSTypeConverter:
+        """Get the default type converter for S3FS cursor.
+
+        Args:
+            unload: Unused. S3FS cursor does not support UNLOAD operations.
+
+        Returns:
+            DefaultS3FSTypeConverter instance.
+        """
+        return DefaultS3FSTypeConverter()
+
+    @property
+    def arraysize(self) -> int:
+        """Get the number of rows to fetch at a time."""
+        return self._arraysize
+
+    @arraysize.setter
+    def arraysize(self, value: int) -> None:
+        """Set the number of rows to fetch at a time.
+
+        Args:
+            value: Number of rows (must be positive).
+
+        Raises:
+            ProgrammingError: If value is not positive.
+        """
+        if value <= 0:
+            raise ProgrammingError("arraysize must be a positive integer value.")
+        self._arraysize = value
+
+    def _collect_result_set(
+        self,
+        query_id: str,
+        kwargs: Optional[Dict[str, Any]] = None,
+    ) -> AthenaS3FSResultSet:
+        """Collect result set after query execution.
+
+        Args:
+            query_id: The Athena query execution ID.
+            kwargs: Additional keyword arguments for result set.
+
+        Returns:
+            AthenaS3FSResultSet containing the query results.
+        """
+        if kwargs is None:
+            kwargs = {}
+        query_execution = cast(AthenaQueryExecution, self._poll(query_id))
+        return AthenaS3FSResultSet(
+            connection=self._connection,
+            converter=self._converter,
+            query_execution=query_execution,
+            arraysize=self._arraysize,
+            retry_config=self._retry_config,
+            **kwargs,
+        )
+
+    def execute(
+        self,
+        operation: str,
+        parameters: Optional[Union[Dict[str, Any], List[str]]] = None,
+        work_group: Optional[str] = None,
+        s3_staging_dir: Optional[str] = None,
+        cache_size: Optional[int] = 0,
+        cache_expiration_time: Optional[int] = 0,
+        result_reuse_enable: Optional[bool] = None,
+        result_reuse_minutes: Optional[int] = None,
+        paramstyle: Optional[str] = None,
+        **kwargs,
+    ) -> Tuple[str, "Future[Union[AthenaS3FSResultSet, Any]]"]:
+        """Execute a SQL query asynchronously.
+
+        Submits the query to Athena and returns immediately with a query ID
+        and a Future that will contain the result set when complete.
+
+        Args:
+            operation: SQL query string to execute.
+            parameters: Query parameters for parameterized queries.
+            work_group: Athena workgroup to use for this query.
+            s3_staging_dir: S3 location for query results.
+            cache_size: Number of queries to check for result caching.
+            cache_expiration_time: Cache expiration time in seconds.
+            result_reuse_enable: Enable Athena result reuse for this query.
+            result_reuse_minutes: Minutes to reuse cached results.
+            paramstyle: Parameter style ('qmark' or 'pyformat').
+            **kwargs: Additional execution parameters.
+
+        Returns:
+            Tuple of (query_id, Future[AthenaS3FSResultSet]).
+
+        Example:
+            >>> query_id, future = cursor.execute("SELECT * FROM my_table")
+            >>> result_set = future.result()
+            >>> rows = result_set.fetchall()
+        """
+        query_id = self._execute(
+            operation,
+            parameters=parameters,
+            work_group=work_group,
+            s3_staging_dir=s3_staging_dir,
+            cache_size=cache_size,
+            cache_expiration_time=cache_expiration_time,
+            result_reuse_enable=result_reuse_enable,
+            result_reuse_minutes=result_reuse_minutes,
+            paramstyle=paramstyle,
+        )
+        return (
+            query_id,
+            self._executor.submit(
+                self._collect_result_set,
+                query_id,
+                kwargs,
+            ),
+        )

pyathena/s3fs/converter.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import logging
+from copy import deepcopy
+from typing import Any, Optional
+
+from pyathena.converter import (
+    _DEFAULT_CONVERTERS,
+    Converter,
+    _to_default,
+)
+
+_logger = logging.getLogger(__name__)
+
+
+class DefaultS3FSTypeConverter(Converter):
+    """Type converter for S3FS Cursor results.
+
+    This converter is specifically designed for the S3FSCursor and provides
+    type conversion for CSV-based result files read via the S3 FileSystem.
+    It converts Athena data types to Python types using the standard
+    converter mappings.
+
+    The converter uses the same mappings as DefaultTypeConverter, providing
+    consistent behavior with the standard Cursor while using the S3FileSystem
+    for file access.
+
+    Example:
+        >>> from pyathena.s3fs.converter import DefaultS3FSTypeConverter
+        >>> converter = DefaultS3FSTypeConverter()
+        >>>
+        >>> # Used automatically by S3FSCursor
+        >>> cursor = connection.cursor(S3FSCursor)
+        >>> # converter is applied automatically to results
+
+    Note:
+        This converter is used by default in S3FSCursor.
+        Most users don't need to instantiate it directly.
+    """
+
+    def __init__(self) -> None:
+        super().__init__(
+            mappings=deepcopy(_DEFAULT_CONVERTERS),
+            default=_to_default,
+        )
+
+    def convert(self, type_: str, value: Optional[str]) -> Optional[Any]:
+        converter = self.get(type_)
+        return converter(value)