""" TiDB Time Decoder - Handles TiDB packed time format Based on TiDB's time.go FromPackedUint and ToPackedUint functions """ from datetime import datetime from typing import Optional def from_packed_uint(packed_uint64: int) -> Optional[datetime]: """ Decode TiDB packed uint64 time format to datetime Based on TiDB's FromPackedUint function (tidb_refs/time.go:659-681) Format: - 1 bit: 0 (unused) - 17 bits: year*13+month (year 0-9999, month 0-12) - 5 bits: day (0-31) - 5 bits: hour (0-23) - 6 bits: minute (0-59) - 6 bits: second (0-59) - 24 bits: microseconds (0-999999) Args: packed_uint64: TiDB packed time value Returns: datetime object or None if packed value is 0 """ if packed_uint64 == 0: return None # Extract components according to TiDB's algorithm ymdhms = packed_uint64 >> 24 ymd = ymdhms >> 17 day = int(ymd & ((1 << 5) - 1)) ym = ymd >> 5 month = int(ym % 13) year = int(ym // 13) hms = ymdhms & ((1 << 17) - 1) second = int(hms & ((1 << 6) - 1)) minute = int((hms >> 6) & ((1 << 6) - 1)) hour = int(hms >> 12) microsec = int(packed_uint64 & ((1 << 24) - 1)) # Normalize components to valid ranges (TiDB may have overflow/precision issues) year = max(1, min(9999, year)) month = max(1, min(12, month if month > 0 else 1)) day = max(1, min(31, day if day > 0 else 1)) # Handle time component overflow (common in TiDB packed time) if hour >= 24: # Add overflow days and normalize hour extra_days = hour // 24 hour = hour % 24 day = min(31, day + extra_days) # Rough overflow handling hour = max(0, min(23, hour)) minute = max(0, min(59, minute)) second = max(0, min(59, second)) # Handle microsecond overflow if microsec >= 1000000: # Add overflow seconds and normalize microseconds extra_seconds = microsec // 1000000 microsec = microsec % 1000000 second = min(59, second + extra_seconds) microsec = max(0, min(999999, microsec)) try: return datetime(year, month, day, hour, minute, second, microsec) except ValueError: # Final fallback: return a reasonable default datetime # This handles edge cases where even normalized components are invalid return datetime(year, month, min(28, day), hour, minute, second, microsec) def to_packed_uint(dt: datetime) -> int: """ Encode datetime to TiDB packed uint64 format Based on TiDB's ToPackedUint function (tidb_refs/time.go:646-657) Args: dt: datetime object to encode Returns: TiDB packed uint64 value """ if dt is None: return 0 year, month, day = dt.year, dt.month, dt.day hour, minute, second = dt.hour, dt.minute, dt.second microsec = dt.microsecond ymd = ((year * 13 + month) << 5) | day hms = hour << 12 | minute << 6 | second return ((ymd << 17 | hms) << 24) | microsec def is_tidb_packed_time(value: int) -> bool: """ Check if an integer value looks like a TiDB packed time format Args: value: Integer value to check Returns: True if value appears to be TiDB packed time format """ if not isinstance(value, int) or value < 0: return False # Very large values (> 1e15) are likely TiDB packed time # Normal Unix timestamps are much smaller if value > 1e15: try: # Try to decode - if it succeeds without major errors, it's likely packed time dt = from_packed_uint(value) # Accept any datetime that was successfully created, even if components were normalized return dt is not None and dt.year >= 1 and dt.year <= 9999 except Exception: return False return False def tidb_time_to_scalar(dt: datetime, data_type: str = "DATETIME") -> float: """ Convert datetime to TiDB-compatible scalar value For TiDB time types, we return the packed uint64 value directly as the scalar, rather than calculating duration from baseline. Args: dt: datetime object data_type: TiDB data type (DATETIME, TIMESTAMP, DATE, TIME) Returns: Scalar value (packed uint64 as float) """ if dt is None: return 0.0 # For TiDB time types, use packed uint64 as scalar packed_value = to_packed_uint(dt) return float(packed_value) def scalar_to_tidb_time( scalar: float, data_type: str = "DATETIME" ) -> Optional[datetime]: """ Convert scalar value back to datetime Args: scalar: Scalar value (should be TiDB packed uint64) data_type: TiDB data type (DATETIME, TIMESTAMP, DATE, TIME) Returns: datetime object or None """ if scalar == 0.0: return None # Convert float scalar back to int (packed uint64) try: packed_value = int(scalar) return from_packed_uint(packed_value) except (ValueError, OverflowError) as e: raise ValueError(f"Failed to convert scalar {scalar} to TiDB time: {e}")