init

Author: millerjp

libs/async-cassandra-dataframe/SPLIT_STRATEGY_USAGE.md

@@ -0,0 +1,92 @@
+# SPLIT Partitioning Strategy
+
+The SPLIT strategy provides manual control over Dask partition count by splitting each Cassandra token range into N sub-partitions.
+
+## When to Use
+
+Use the SPLIT strategy when:
+- Automatic partition calculations are too conservative
+- You need more parallelism for large datasets
+- Token ranges contain uneven data distribution
+- You want fine-grained control over partition count
+
+## Usage
+
+```python
+import async_cassandra_dataframe as cdf
+
+# Split each token range into 3 sub-partitions
+df = await cdf.read_cassandra_table(
+    "my_table",
+    session=session,
+    partitioning_strategy="split",  # Use SPLIT strategy
+    split_factor=3,                  # Split each range into 3
+)
+
+# Example: 17 token ranges * 3 splits = 51 Dask partitions
+```
+
+## How It Works
+
+1. Discovers natural token ranges from Cassandra cluster
+2. Splits each token range into N equal sub-ranges
+3. Creates one Dask partition per sub-range
+
+## Examples
+
+### Basic Usage
+```python
+# Default AUTO strategy (conservative)
+df_auto = await cdf.read_cassandra_table("my_table", session=session)
+# Result: 2 partitions for medium dataset
+
+# SPLIT strategy with factor 5
+df_split = await cdf.read_cassandra_table(
+    "my_table",
+    session=session,
+    partitioning_strategy="split",
+    split_factor=5,
+)
+# Result: 85 partitions (17 ranges * 5)
+```
+
+### High Parallelism
+```python
+# For CPU-intensive processing, increase parallelism
+df = await cdf.read_cassandra_table(
+    "large_table",
+    session=session,
+    partitioning_strategy="split",
+    split_factor=10,  # 10x more partitions
+)
+
+# Process with Dask
+result = df.map_partitions(expensive_computation).compute()
+```
+
+### Comparison with Other Strategies
+
+| Strategy | Use Case | Partition Count |
+|----------|----------|-----------------|
+| AUTO | General purpose | Conservative (2-10) |
+| NATURAL | Maximum parallelism | One per token range |
+| COMPACT | Memory-bounded | Based on target size |
+| FIXED | Specific count | User-specified |
+| SPLIT | Manual control | Token ranges * split_factor |
+
+## Performance Considerations
+
+- Higher split_factor = more parallelism but also more overhead
+- Each partition requires a separate Cassandra query
+- Optimal split_factor depends on:
+  - Data volume per token range
+  - Available CPU cores
+  - Processing complexity
+  - Network latency
+
+## Recommendations
+
+- Start with split_factor=2-5 for most cases
+- Use 10+ for CPU-intensive processing on large clusters
+- Monitor partition sizes with logging
+- Adjust based on performance measurements

libs/async-cassandra-dataframe/src/async_cassandra_dataframe/partition_strategy.py

@@ -23,6 +23,7 @@ class PartitioningStrategy(str, Enum):
     NATURAL = "natural"  # One partition per token range
     COMPACT = "compact"  # Balance parallelism and overhead
     FIXED = "fixed"  # User-specified partition count
+    SPLIT = "split"  # Split each token range into N sub-partitions
 
 
 @dataclass
@@ -70,6 +71,7 @@ def group_token_ranges(
         strategy: PartitioningStrategy = PartitioningStrategy.AUTO,
         target_partition_count: int | None = None,
         target_partition_size_mb: int | None = None,
+        split_factor: int | None = None,
     ) -> list[PartitionGroup]:
         """
         Group token ranges into partitions based on strategy.
@@ -79,6 +81,7 @@ def group_token_ranges(
             strategy: Partitioning strategy to use
             target_partition_count: Desired number of partitions (for FIXED strategy)
             target_partition_size_mb: Target size per partition
+            split_factor: Number of sub-partitions per token range (for SPLIT strategy)
 
         Returns:
             List of partition groups
@@ -96,6 +99,10 @@ def group_token_ranges(
             if target_partition_count is None:
                 raise ValueError("FIXED strategy requires target_partition_count")
             return self._fixed_grouping(token_ranges, target_partition_count)
+        elif strategy == PartitioningStrategy.SPLIT:
+            if split_factor is None:
+                raise ValueError("SPLIT strategy requires split_factor")
+            return self._split_grouping(token_ranges, split_factor)
         else:  # AUTO
             return self._auto_grouping(token_ranges, target_size)
 
@@ -260,6 +267,47 @@ def _auto_grouping(
                 target_partitions = max(len(token_ranges) // 2, unique_nodes * 4)
                 return self._fixed_grouping(token_ranges, target_partitions)
 
+    def _split_grouping(
+        self, token_ranges: list[TokenRange], split_factor: int
+    ) -> list[PartitionGroup]:
+        """
+        Split each token range into N sub-partitions.
+
+        Args:
+            token_ranges: Original token ranges from Cassandra
+            split_factor: Number of sub-partitions per token range
+
+        Returns:
+            List of partition groups, one per sub-range
+        """
+        groups = []
+        partition_id = 0
+
+        for token_range in token_ranges:
+            # Split the token range into sub-ranges
+            sub_ranges = token_range.split(split_factor)
+
+            # Create a partition group for each sub-range
+            for sub_range in sub_ranges:
+                # Estimate size based on fraction
+                estimated_size = self.default_partition_size_mb * sub_range.fraction
+
+                group = PartitionGroup(
+                    partition_id=partition_id,
+                    token_ranges=[sub_range],
+                    estimated_size_mb=estimated_size,
+                    primary_replica=sub_range.replicas[0] if sub_range.replicas else None,
+                )
+                groups.append(group)
+                partition_id += 1
+
+        logger.info(
+            f"Split partitioning: {len(token_ranges)} ranges split by {split_factor} "
+            f"= {len(groups)} partitions"
+        )
+
+        return groups
+
     def _group_by_replica(self, token_ranges: list[TokenRange]) -> dict[str, list[TokenRange]]:
         """Group token ranges by their primary replica."""
         ranges_by_replica: dict[str, list[TokenRange]] = {}

libs/async-cassandra-dataframe/src/async_cassandra_dataframe/reader.py

@@ -172,6 +172,7 @@ async def read(
         # Partitioning strategy
         partition_strategy: str = "auto",
         target_partition_size_mb: int = 1024,
+        split_factor: int | None = None,
         # Validation
         require_partition_key_predicate: bool = False,
         # Progress
@@ -224,6 +225,7 @@ async def read(
             pushdown_predicates,
             partition_strategy,
             target_partition_size_mb,
+            split_factor,
         )
 
         # Normalize snapshot time
@@ -454,6 +456,7 @@ async def _create_partitions(
         pushdown_predicates: list,
         partition_strategy: str,
         target_partition_size_mb: int,
+        split_factor: int | None,
     ) -> list[dict[str, Any]]:
         """Create partition definitions."""
         # Create partition strategy
@@ -482,6 +485,7 @@ async def _create_partitions(
                     columns,
                     None,  # writetime_columns
                     None,  # ttl_columns
+                    split_factor,
                 )
             except Exception as e:
                 logger.warning(f"Could not apply partitioning strategy: {e}")
@@ -497,6 +501,7 @@ async def _create_grouped_partitions(
         columns: list[str],
         writetime_columns: list[str] | None,
         ttl_columns: list[str] | None,
+        split_factor: int | None,
     ) -> list[dict[str, Any]]:
         """Create grouped partitions based on partitioning strategy."""
         # Get natural token ranges
@@ -513,6 +518,7 @@ async def _create_grouped_partitions(
             strategy=strategy_enum,
             target_partition_count=partition_count,
             target_partition_size_mb=target_partition_size_mb,
+            split_factor=split_factor,
         )
 
         # Log partitioning info
@@ -650,7 +656,9 @@ async def read_cassandra_table(
     adaptive_page_size: bool = False,
     # Partitioning strategy
     partition_strategy: str = "auto",
+    partitioning_strategy: str | None = None,  # Alias for backward compatibility
     target_partition_size_mb: int = 1024,
+    split_factor: int | None = None,
     # Validation
     require_partition_key_predicate: bool = False,
     # Progress
@@ -687,8 +695,9 @@ async def read_cassandra_table(
         max_concurrent_partitions=max_concurrent_partitions,
         page_size=page_size,
         adaptive_page_size=adaptive_page_size,
-        partition_strategy=partition_strategy,
+        partition_strategy=partitioning_strategy or partition_strategy,  # Use alias if provided
         target_partition_size_mb=target_partition_size_mb,
+        split_factor=split_factor,
         require_partition_key_predicate=require_partition_key_predicate,
         progress_callback=progress_callback,
         client=client,

libs/async-cassandra-dataframe/src/async_cassandra_dataframe/token_ranges.py

@@ -63,6 +63,68 @@ def contains_token(self, token: int) -> bool:
             # Wraparound: token is either after start OR before end
             return token >= self.start or token <= self.end
 
+    def split(self, split_factor: int) -> list["TokenRange"]:
+        """
+        Split this token range into N equal sub-ranges.
+
+        Args:
+            split_factor: Number of sub-ranges to create
+
+        Returns:
+            List of sub-ranges that cover this range
+
+        Raises:
+            ValueError: If split_factor is not positive
+        """
+        if split_factor < 1:
+            raise ValueError("split_factor must be positive")
+
+        if split_factor == 1:
+            return [self]
+
+        # Handle wraparound ranges
+        if self.is_wraparound:
+            # Split into two non-wraparound ranges first
+            first_part = TokenRange(start=self.start, end=MAX_TOKEN, replicas=self.replicas)
+            second_part = TokenRange(start=MIN_TOKEN, end=self.end, replicas=self.replicas)
+
+            # Calculate how to distribute splits between the two parts
+            first_size = first_part.size
+            second_size = second_part.size
+            total_size = first_size + second_size
+
+            # Allocate splits proportionally
+            first_splits = max(1, round(split_factor * first_size / total_size))
+            second_splits = max(1, split_factor - first_splits)
+
+            result = []
+            result.extend(first_part.split(first_splits))
+            result.extend(second_part.split(second_splits))
+            return result
+
+        # Calculate split size
+        range_size = self.size
+        if range_size < split_factor:
+            # Can't split into more parts than tokens available
+            # Still create the requested number of splits, some may be very small
+            pass
+
+        splits = []
+        for i in range(split_factor):
+            # Calculate boundaries for this split
+            if i == split_factor - 1:
+                # Last split gets any remainder
+                start = self.start + (range_size * i // split_factor)
+                end = self.end
+            else:
+                start = self.start + (range_size * i // split_factor)
+                end = self.start + (range_size * (i + 1) // split_factor)
+
+            # Create sub-range with proportional fraction
+            splits.append(TokenRange(start=start, end=end, replicas=self.replicas))
+
+        return splits
+
 
 async def discover_token_ranges(session: Any, keyspace: str) -> list[TokenRange]:
     """