improve get_smallest_cron_interval

Author: anuthebananu

python_modules/dagster/dagster/_utils/schedules.py

@@ -866,13 +866,166 @@ def get_smallest_cron_interval(
     cron_string: str,
     execution_timezone: Optional[str] = None,
 ) -> datetime.timedelta:
-    """Find the smallest interval between cron ticks for a given cron schedule.
+    """Find the smallest interval between cron ticks for a given cron schedule using deterministic
+    analysis of the cron pattern.
 
-    Uses a sampling-based approach to find the minimum interval by generating
+    This function parses the cron string and algebraically determines the minimum interval without
+    sampling. This is more efficient and deterministic than get_smallest_cron_interval() for most
+    common patterns.
+
+    For complex patterns that cannot be analyzed deterministically (e.g., patterns with both
+    day-of-month AND day-of-week constraints, or irregular intervals), this falls back to the
+    sampling-based approach.
+
+    Args:
+        cron_string: A cron string
+        execution_timezone: Timezone to use for cron evaluation (only used for fallback)
+
+    Returns:
+        The smallest timedelta between any two consecutive cron ticks
+
+    Raises:
+        CheckError: If the cron string is invalid or not recognized by Dagster
+    """
+    check.invariant(
+        is_valid_cron_string(cron_string), desc=f"{cron_string} must be a valid cron string"
+    )
+
+    # Parse the cron string into its components: [minutes, hours, day_of_month, month, day_of_week]
+    # Each component is a list of int or '*'
+    cron_parts, nth_weekday_of_month, *_ = CroniterShim.expand(cron_string)
+
+    # If nth_weekday_of_month is used (e.g., "first Monday of the month"), fall back to sampling
+    if nth_weekday_of_month:
+        return _get_smallest_cron_interval_with_sampling(cron_string, execution_timezone)
+
+    minutes, hours, days_of_month, months, days_of_week = cron_parts
+
+    # Helper function to get smallest gap in a sorted list of integers
+    def get_smallest_gap(values: list[int], wrap_at: Optional[int] = None) -> Optional[int]:
+        """Get the smallest gap between consecutive values in a sorted list.
+
+        Args:
+            values: List of integer values
+            wrap_at: If provided, also considers wrap-around gap (e.g., 60 for minutes)
+        """
+        if len(values) < 2:
+            return None
+        sorted_values = sorted(values)
+
+        # Calculate gaps between consecutive values
+        gaps = [sorted_values[i + 1] - sorted_values[i] for i in range(len(sorted_values) - 1)]
+
+        # If wrap_at is provided, also consider the wrap-around gap
+        if wrap_at is not None:
+            wrap_gap = (wrap_at - sorted_values[-1]) + sorted_values[0]
+            gaps.append(wrap_gap)
+
+        return min(gaps)
+
+    # Determine if each field is constrained or wildcarded
+    minutes_is_wildcard = len(minutes) == 1 and minutes[0] == "*"
+    hours_is_wildcard = len(hours) == 1 and hours[0] == "*"
+    days_of_month_is_wildcard = len(days_of_month) == 1 and days_of_month[0] == "*"
+    months_is_wildcard = len(months) == 1 and months[0] == "*"
+    days_of_week_is_wildcard = len(days_of_week) == 1 and days_of_week[0] == "*"
+
+    # If both day_of_month and day_of_week are constrained, they use OR logic which is complex
+    # Fall back to sampling for these cases
+    if not days_of_month_is_wildcard and not days_of_week_is_wildcard:
+        return _get_smallest_cron_interval_with_sampling(cron_string, execution_timezone)
+
+    # Extract numeric values (filter out '*')
+    minute_values = [m for m in minutes if m != "*"]
+    hour_values = [h for h in hours if h != "*"]
+    day_of_week_values = [d for d in days_of_week if d != "*"]
+
+    # Case 1: Minutes are wildcarded (* in minutes position)
+    # This means the job runs every minute during the matching hours
+    if minutes_is_wildcard:
+        return datetime.timedelta(minutes=1)
+
+    # Case 2: Multiple minute values specified (e.g., "0,15,30,45")
+    # The smallest interval is the minimum gap between minute values
+    if len(minute_values) > 1:
+        min_minute_gap = get_smallest_gap(minute_values, wrap_at=60)
+        if min_minute_gap is not None:
+            # If hours/days/months/weekdays are all wildcarded, this is the answer
+            if (
+                hours_is_wildcard
+                and days_of_month_is_wildcard
+                and months_is_wildcard
+                and days_of_week_is_wildcard
+            ):
+                return datetime.timedelta(minutes=min_minute_gap)
+            # Otherwise, we need to consider if the time constraints might make consecutive ticks
+            # happen at different hours/days. This is complex, so fall back to sampling.
+            return _get_smallest_cron_interval_with_sampling(cron_string, execution_timezone)
+
+    # Case 3: Single minute value specified (e.g., "0" or "15")
+    # Now we need to look at the hour constraints
+    if len(minute_values) == 1:
+        # If hours are wildcarded, runs every hour at that minute
+        if hours_is_wildcard:
+            # Check day/month/week constraints
+            if days_of_month_is_wildcard and months_is_wildcard and days_of_week_is_wildcard:
+                return datetime.timedelta(hours=1)
+            # If days/months/weeks are constrained, fall back to sampling
+            return _get_smallest_cron_interval_with_sampling(cron_string, execution_timezone)
+
+        # Multiple hour values specified
+        if len(hour_values) > 1:
+            min_hour_gap = get_smallest_gap(hour_values, wrap_at=24)
+            if min_hour_gap is not None:
+                # If days/months/weekdays are all wildcarded, the interval is based on hours
+                if days_of_month_is_wildcard and months_is_wildcard and days_of_week_is_wildcard:
+                    return datetime.timedelta(hours=min_hour_gap)
+                # Otherwise, constraints might make it more complex
+                return _get_smallest_cron_interval_with_sampling(cron_string, execution_timezone)
+
+        # Single hour value specified (e.g., "0 0 * * *" - daily at midnight)
+        if len(hour_values) == 1:
+            # Daily pattern: specific minute and hour, all days
+            if days_of_month_is_wildcard and months_is_wildcard and days_of_week_is_wildcard:
+                return datetime.timedelta(days=1)
+
+            # Weekly pattern: specific minute, hour, and day of week
+            if days_of_month_is_wildcard and months_is_wildcard and len(day_of_week_values) == 1:
+                return datetime.timedelta(days=7)
+
+            # Multiple days of week (e.g., Mon, Wed, Fri)
+            if days_of_month_is_wildcard and months_is_wildcard and len(day_of_week_values) > 1:
+                min_dow_gap = get_smallest_gap(day_of_week_values, wrap_at=7)
+                if min_dow_gap is not None:
+                    return datetime.timedelta(days=min_dow_gap)
+
+            # Monthly pattern: specific day of month
+            if not days_of_month_is_wildcard and months_is_wildcard and days_of_week_is_wildcard:
+                # For monthly patterns, the interval varies (28-31 days depending on the month)
+                # Fall back to sampling for accuracy
+                return _get_smallest_cron_interval_with_sampling(cron_string, execution_timezone)
+
+            # Complex pattern with month constraints
+            if not months_is_wildcard:
+                return _get_smallest_cron_interval_with_sampling(cron_string, execution_timezone)
+
+    # If we haven't returned yet, fall back to sampling-based approach
+    return _get_smallest_cron_interval_with_sampling(cron_string, execution_timezone)
+
+
+def _get_smallest_cron_interval_with_sampling(
+    cron_string: str,
+    execution_timezone: Optional[str] = None,
+) -> datetime.timedelta:
+    """Find the smallest interval between cron ticks for a given cron schedule,
+    using a sampling-based approach to find the minimum interval by generating
     consecutive cron ticks and measuring the gaps between them. Sampling stops
     early if either of these limits is reached:
       - A maximum of 1000 generated ticks
-      - A time horizon of 20 years past the sampling start
+      - A time horizon of 20 years past the sampling start.
+
+    This is a fallback for complex patterns that cannot be analyzed deterministically,
+    and shouldn't be used for common patterns.
 
     Args:
         cron_string: A cron string
@@ -937,6 +1090,12 @@ def get_smallest_cron_interval(
                 # We've encountered a genuine zero interval (which shouldn't happen)
                 raise Exception("Encountered a genuine zero interval")
 
+            if interval < datetime.timedelta(seconds=0):
+                # This happens when the sampling encounters a daylight savings transition where the clocks roll back
+                # Just skip this interval and continue sampling
+                prev_tick = current_tick
+                continue
+
             # Update minimum interval
             if min_interval is None or interval < min_interval:
                 min_interval = interval

python_modules/dagster/dagster_tests/scheduler_tests/test_cron_string_iterator.py

@@ -6,11 +6,9 @@
 from dagster._utils.schedules import (
     _croniter_string_iterator,
     cron_string_iterator,
-    get_smallest_cron_interval,
     is_valid_cron_string,
     reverse_cron_string_iterator,
 )
-from dagster_shared.check import CheckError
 
 
 def test_cron_iterator_always_advances():
@@ -678,148 +676,3 @@ def test_invalid_cron_strings():
 
     assert is_valid_cron_string("0 0 31 1 *")
     assert not is_valid_cron_string("0 0 32 1 *")
-
-
-def test_get_smallest_cron_interval_basic():
-    """Test basic cron intervals return expected minimums."""
-    # Minute intervals
-    assert get_smallest_cron_interval("*/5 * * * *") == datetime.timedelta(minutes=5)
-    assert get_smallest_cron_interval("*/15 * * * *") == datetime.timedelta(minutes=15)
-    assert get_smallest_cron_interval("*/30 * * * *") == datetime.timedelta(minutes=30)
-
-    # Hourly intervals
-    assert get_smallest_cron_interval("0 * * * *") == datetime.timedelta(hours=1)
-    assert get_smallest_cron_interval("0 */6 * * *") == datetime.timedelta(hours=6)
-    assert get_smallest_cron_interval("0 */12 * * *") == datetime.timedelta(hours=12)
-
-    # Daily intervals
-    assert get_smallest_cron_interval("0 0 * * *") == datetime.timedelta(days=1)
-    assert get_smallest_cron_interval("30 14 * * *") == datetime.timedelta(days=1)
-
-    # Weekly intervals
-    assert get_smallest_cron_interval("0 0 * * 0") == datetime.timedelta(days=7)
-    assert get_smallest_cron_interval("0 9 * * 1") == datetime.timedelta(days=7)
-
-
-def test_get_smallest_cron_interval_irregular():
-    """Test irregular cron schedules return correct minimum intervals."""
-    # Multiple times per hour
-    interval = get_smallest_cron_interval("15,45 * * * *")
-    assert interval == datetime.timedelta(minutes=30)
-
-    # Multiple times per day
-    interval = get_smallest_cron_interval("0 9,17 * * *")
-    assert interval == datetime.timedelta(hours=8)
-
-    # Weekdays only
-    interval = get_smallest_cron_interval("0 9 * * 1-5")
-    assert interval == datetime.timedelta(days=1)  # Daily on weekdays
-
-    # Multiple days per week
-    interval = get_smallest_cron_interval("0 9 * * 1,3,5")
-    assert interval == datetime.timedelta(days=2)  # Mon->Wed->Fri pattern
-
-
-def test_get_smallest_cron_interval_monthly():
-    """Test monthly cron schedules."""
-    # Monthly on 1st
-    interval = get_smallest_cron_interval("0 0 1 * *")
-    # Shortest month interval is 28 days (February)
-    assert interval == datetime.timedelta(days=28)
-
-    # Monthly on 15th
-    interval = get_smallest_cron_interval("0 12 15 * *")
-    assert interval == datetime.timedelta(days=28)
-
-
-def test_get_smallest_cron_interval_leap_year():
-    """Test leap year edge case with Feb 29th."""
-    # Feb 29th only runs on leap years
-    interval = get_smallest_cron_interval("0 0 29 2 *")
-    # Should be 1 year for non-leap years, but our sampling should catch 4-year intervals
-    # during leap year sequences
-    assert interval.days >= 365  # At least 1 year
-
-    # The exact value depends on when we sample, but should be reasonable
-    assert interval.days <= 4 * 365 + 1  # At most 4 years + leap day
-
-
-def test_get_smallest_cron_interval_dst_transitions():
-    """Test DST transition edge cases."""
-    # Daily at 2am in a DST timezone - should catch the 23-hour interval during spring forward
-    interval = get_smallest_cron_interval("0 2 * * *", "America/New_York")
-    assert interval == datetime.timedelta(hours=23)
-
-    # Hourly schedule should not be affected by DST for minimum interval
-    interval = get_smallest_cron_interval("0 * * * *", "America/New_York")
-    assert interval == datetime.timedelta(hours=1)
-
-    # Different timezone with DST
-    interval = get_smallest_cron_interval("0 2 * * *", "Europe/Berlin")
-    assert interval == datetime.timedelta(hours=23)
-
-
-def test_get_smallest_cron_interval_timezones():
-    """Test various timezones work correctly."""
-    # UTC should work
-    interval = get_smallest_cron_interval("*/10 * * * *", "UTC")
-    assert interval == datetime.timedelta(minutes=10)
-
-    # Other timezones should work
-    interval = get_smallest_cron_interval("*/10 * * * *", "Asia/Tokyo")
-    assert interval == datetime.timedelta(minutes=10)
-
-    # Default timezone (UTC) should work
-    interval = get_smallest_cron_interval("*/10 * * * *")
-    assert interval == datetime.timedelta(minutes=10)
-
-
-def test_get_smallest_cron_interval_complex_patterns():
-    """Test complex cron patterns."""
-    # Every 5 minutes during business hours on weekdays
-    interval = get_smallest_cron_interval("*/5 9-17 * * 1-5")
-    assert interval == datetime.timedelta(minutes=5)
-
-    # Multiple specific times
-    interval = get_smallest_cron_interval("0,30 8,12,16 * * 1-5")
-    assert interval == datetime.timedelta(minutes=30)
-
-    # Specific day patterns
-    interval = get_smallest_cron_interval("0 9 1,15 * *")  # 1st and 15th of month
-    # Minimum should be 14 days (15th to 1st of next month can be 14-17 days)
-    assert interval.days >= 14
-    assert interval.days <= 17
-
-
-def test_get_smallest_cron_interval_edge_cases():
-    """Test edge cases and error conditions."""
-    # Invalid cron string should raise error
-    with pytest.raises(CheckError):
-        get_smallest_cron_interval("invalid cron")
-
-    with pytest.raises(CheckError):
-        get_smallest_cron_interval("0 0 32 * *")  # Invalid day
-
-    # Valid but unusual patterns
-    interval = get_smallest_cron_interval("0 0 * * *")  # Daily
-    assert interval == datetime.timedelta(days=1)
-
-    # Very frequent pattern
-    interval = get_smallest_cron_interval("* * * * *")  # Every minute
-    assert interval == datetime.timedelta(minutes=1)
-
-
-def test_get_smallest_cron_interval_consistency():
-    """Test that the method returns consistent results."""
-    # Run the same cron string multiple times to ensure consistency
-    cron_string = "*/15 * * * *"
-    timezone = "America/Los_Angeles"
-
-    results = []
-    for _ in range(3):
-        interval = get_smallest_cron_interval(cron_string, timezone)
-        results.append(interval)
-
-    # All results should be the same
-    assert all(result == results[0] for result in results)
-    assert results[0] == datetime.timedelta(minutes=15)