paimon-python/pypaimon/common/time_utils.py - paimon - Git at Google

 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
 # distributed with this work for additional information
 # regarding copyright ownership.  The ASF licenses this file
 # to you under the Apache License, Version 2.0 (the
 # "License"); you may not use this file except in compliance
 # with the License.  You may obtain a copy of the License at
 #
 #   http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing,
 # software distributed under the License is distributed on an
 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.

 import calendar
 from datetime import datetime, timedelta
 from typing import List


 def parse_duration(text: str) -> int:
     if text is None:
         raise ValueError("text cannot be None")

     trimmed = text.strip().lower()
     if not trimmed:
         raise ValueError("argument is an empty- or whitespace-only string")

     pos = 0
     while pos < len(trimmed) and trimmed[pos].isdigit():
         pos += 1

     number_str = trimmed[:pos]
     unit_str = trimmed[pos:].strip()

     if not number_str:
         raise ValueError("text does not start with a number")

     try:
         value = int(number_str)
     except ValueError:
         raise ValueError(
             f"The value '{number_str}' cannot be re represented as 64bit number (numeric overflow)."
         )

     if not unit_str:
         result_ms = value
     elif unit_str in ('ns', 'nano', 'nanosecond', 'nanoseconds'):
         result_ms = value / 1_000_000
     elif unit_str in ('µs', 'micro', 'microsecond', 'microseconds'):
         result_ms = value / 1_000
     elif unit_str in ('ms', 'milli', 'millisecond', 'milliseconds'):
         result_ms = value
     elif unit_str in ('s', 'sec', 'second', 'seconds'):
         result_ms = value * 1000
     elif unit_str in ('m', 'min', 'minute', 'minutes'):
         result_ms = value * 60 * 1000
     elif unit_str in ('h', 'hour', 'hours'):
         result_ms = value * 60 * 60 * 1000
     elif unit_str in ('d', 'day', 'days'):
         result_ms = value * 24 * 60 * 60 * 1000
     else:
         supported_units = (
             'DAYS: (d | day | days), '
             'HOURS: (h | hour | hours), '
             'MINUTES: (m | min | minute | minutes), '
             'SECONDS: (s | sec | second | seconds), '
             'MILLISECONDS: (ms | milli | millisecond | milliseconds), '
             'MICROSECONDS: (µs | micro | microsecond | microseconds), '
             'NANOSECONDS: (ns | nano | nanosecond | nanoseconds)'
         )
         raise ValueError(
             f"Time interval unit label '{unit_str}' does not match any of the recognized units: "
             f"{supported_units}"
         )

     result_ms_int = int(round(result_ms))

     if result_ms_int < 0:
         raise ValueError(f"Duration cannot be negative: {text}")

     return result_ms_int


 # Unit label -> nanoseconds. Mirrors the aliases accepted by ``parse_duration``
 # (and Java ``TimeUtils.parseDuration``); a bare number with no unit is treated
 # as milliseconds, matching ``parse_duration``.
 _UNIT_TO_NANOS = {
     'ns': 1, 'nano': 1, 'nanosecond': 1, 'nanoseconds': 1,
     'µs': 1_000, 'micro': 1_000, 'microsecond': 1_000, 'microseconds': 1_000,
     'ms': 1_000_000, 'milli': 1_000_000, 'millisecond': 1_000_000, 'milliseconds': 1_000_000,
     's': 1_000_000_000, 'sec': 1_000_000_000, 'second': 1_000_000_000, 'seconds': 1_000_000_000,
     'm': 60_000_000_000, 'min': 60_000_000_000, 'minute': 60_000_000_000, 'minutes': 60_000_000_000,
     'h': 3_600_000_000_000, 'hour': 3_600_000_000_000, 'hours': 3_600_000_000_000,
     'd': 86_400_000_000_000, 'day': 86_400_000_000_000, 'days': 86_400_000_000_000,
 }


 def parse_duration_nanos(text: str) -> int:
     """Parse a duration string to an integer nanosecond count.

     This is the full-precision integer counterpart of :func:`parse_duration`
     (which returns rounded milliseconds): it accepts the same unit aliases and
     the same "bare number means milliseconds" convention, but keeps sub-millisecond
     units exactly (``"1ns"`` -> ``1``, ``"500micro"`` -> ``500_000``) instead of
     rounding them to zero. Use it where sub-millisecond precision must not be
     silently dropped. ``parse_duration`` is intentionally left untouched so its
     millisecond contract (relied on by option parsing) does not change.
     """
     if text is None:
         raise ValueError("text cannot be None")

     trimmed = text.strip().lower()
     if not trimmed:
         raise ValueError("argument is an empty- or whitespace-only string")

     pos = 0
     while pos < len(trimmed) and trimmed[pos].isdigit():
         pos += 1

     number_str = trimmed[:pos]
     unit_str = trimmed[pos:].strip()

     if not number_str:
         raise ValueError("text does not start with a number")

     try:
         value = int(number_str)
     except ValueError:
         raise ValueError(
             f"The value '{number_str}' cannot be re represented as 64bit number (numeric overflow)."
         )

     if not unit_str:
         nanos_per_unit = 1_000_000  # bare number is milliseconds
     elif unit_str in _UNIT_TO_NANOS:
         nanos_per_unit = _UNIT_TO_NANOS[unit_str]
     else:
         supported_units = (
             'DAYS: (d | day | days), '
             'HOURS: (h | hour | hours), '
             'MINUTES: (m | min | minute | minutes), '
             'SECONDS: (s | sec | second | seconds), '
             'MILLISECONDS: (ms | milli | millisecond | milliseconds), '
             'MICROSECONDS: (µs | micro | microsecond | microseconds), '
             'NANOSECONDS: (ns | nano | nanosecond | nanoseconds)'
         )
         raise ValueError(
             f"Time interval unit label '{unit_str}' does not match any of the recognized units: "
             f"{supported_units}"
         )

     return value * nanos_per_unit


 # ---------------------------------------------------------------------------
 # Codecs for Java temporal types as serialized by Jackson's JavaTimeModule.
 #
 # These mirror the on-disk JSON shapes that Paimon's Java side reads/writes so
 # that a tag file written by pypaimon is interoperable with Java and vice versa
 # (see ``org.apache.paimon.tag.Tag`` and ``TagTest``):
 #   - ``LocalDateTime`` -> JSON array ``[year, month, day, hour, minute,
 #     second, nanoOfSecond]``
 #   - ``Duration``      -> JSON number of (fractional) seconds, e.g. ``86400.0``
 #
 # Resolution note: Python's ``datetime`` / ``timedelta`` are microsecond-based,
 # so a Java create-time or duration finer than a microsecond is truncated to
 # microseconds on read. Tag granularity is coarse, so this is not a concern in
 # practice.
 # ---------------------------------------------------------------------------

 _NANOS_PER_MICRO = 1000


 def local_datetime_to_json_array(dt: datetime) -> List[int]:
     """Encode a naive ``datetime`` as Java ``LocalDateTime`` array form.

     Matches Jackson's ``LocalDateTimeSerializer`` byte-for-byte: it always emits
     ``[year, month, day, hour, minute]``, appends ``second`` only when second or
     nanoOfSecond is non-zero, and appends ``nanoOfSecond`` only when it is
     non-zero -- i.e. trailing zero components are omitted. ``json_array_to_local_datetime``
     pads them back, so a shorter array round-trips. Python ``datetime`` only has
     microsecond resolution, so the emitted nanoOfSecond is ``microsecond * 1000``
     (never finer than microseconds).
     """
     nano = dt.microsecond * _NANOS_PER_MICRO
     arr = [dt.year, dt.month, dt.day, dt.hour, dt.minute]
     if dt.second != 0 or nano != 0:
         arr.append(dt.second)
         if nano != 0:
             arr.append(nano)
     return arr


 def json_array_to_local_datetime(arr: List[int]) -> datetime:
     """Decode a Java ``LocalDateTime`` array form into a naive ``datetime``.

     Jackson omits trailing zero components, so the array may be shorter than 7;
     missing components default to 0. A Java nanoOfSecond finer than a
     microsecond is truncated to microseconds (Python's resolution limit).
     """
     padded = list(arr) + [0] * (7 - len(arr))
     year, month, day, hour, minute, second, nano = padded[:7]
     return datetime(
         year, month, day, hour, minute, second, nano // _NANOS_PER_MICRO
     )


 def duration_to_json_seconds(td: timedelta):
     """Encode a ``timedelta`` as Java ``Duration`` decimal-seconds number."""
     return td.total_seconds()


 def json_seconds_to_duration(seconds) -> timedelta:
     """Decode a Java ``Duration`` decimal-seconds number into a ``timedelta``."""
     return timedelta(seconds=seconds)


 def duration_to_iso8601(td: timedelta) -> str:
     """Render a non-negative ``timedelta`` like ``java.time.Duration.toString()``.

     Examples: 1 day -> ``PT24H`` (Duration has no day unit), 30 min ->
     ``PT30M``, 5 s -> ``PT5S``, zero -> ``PT0S``. Matches what Paimon's Java
     ``$tags`` system table surfaces for ``time_retained``. Retentions only ever
     come from ``parse_duration``, which rejects negatives, so only the
     non-negative form is supported.
     """
     total_micros = (
         td.days * 86_400_000_000 + td.seconds * 1_000_000 + td.microseconds
     )
     if total_micros <= 0:
         return "PT0S"

     total_seconds, micros = divmod(total_micros, 1_000_000)
     hours, rem = divmod(total_seconds, 3600)
     minutes, secs = divmod(rem, 60)

     buf = "PT"
     if hours != 0:
         buf += "{}H".format(hours)
     if minutes != 0:
         buf += "{}M".format(minutes)
     if secs != 0 or micros != 0 or buf == "PT":
         frac = ""
         if micros != 0:
             frac = "." + ("%06d" % micros).rstrip("0")
         buf += "{}{}S".format(secs, frac)
     return buf


 def local_datetime_to_millis(dt: datetime) -> int:
     """Convert a naive ``LocalDateTime`` to epoch millis, treating it as UTC.

     This is the zone-less / wall-clock conversion: it mirrors Java
     ``Timestamp.fromLocalDateTime`` and feeds the ``$tags`` system table's
     ``create_time`` (a ``TIMESTAMP`` column, which is itself zone-less), so the
     result does NOT depend on the host's time zone. For the REST
     ``GetTagResponse`` epoch-millis ``Long`` use
     :func:`local_datetime_to_system_zone_millis` instead. Uses integer math with
     floored sub-millisecond truncation (matching Java ``Instant.toEpochMilli``),
     so it is exact and correct for pre-epoch instants too.
     """
     return calendar.timegm(dt.timetuple()) * 1000 + dt.microsecond // 1000


 def local_datetime_to_system_zone_millis(dt: datetime) -> int:
     """Convert a naive ``LocalDateTime`` to epoch millis in the host's default
     time zone.

     Mirrors Java ``LocalDateTime.atZone(ZoneId.systemDefault()).toInstant()
     .toEpochMilli()`` -- the conversion the REST ``GetTagResponse`` applies to
     ``tagCreateTime`` (see ``RESTFileSystemCatalog#getTag``). Unlike
     :func:`local_datetime_to_millis` (zone-less / UTC, used by the ``$tags``
     Timestamp column), the result depends on the host's local time zone, so the
     two intentionally differ on non-UTC hosts -- exactly as the two Java paths do.
     """
     # A naive datetime.timestamp() interprets dt in the system local zone (via
     # mktime). Take whole seconds first to avoid float rounding, then floor the
     # sub-second part to millis to match Instant.toEpochMilli().
     epoch_seconds = int(dt.replace(microsecond=0).timestamp())
     return epoch_seconds * 1000 + dt.microsecond // 1000
	# Licensed to the Apache Software Foundation (ASF) under one
	# or more contributor license agreements. See the NOTICE file
	# distributed with this work for additional information
	# regarding copyright ownership. The ASF licenses this file
	# to you under the Apache License, Version 2.0 (the
	# "License"); you may not use this file except in compliance
	# with the License. You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing,
	# software distributed under the License is distributed on an
	# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	# KIND, either express or implied. See the License for the
	# specific language governing permissions and limitations
	# under the License.

	import calendar
	from datetime import datetime, timedelta
	from typing import List


	def parse_duration(text: str) -> int:
	if text is None:
	raise ValueError("text cannot be None")

	trimmed = text.strip().lower()
	if not trimmed:
	raise ValueError("argument is an empty- or whitespace-only string")

	pos = 0
	while pos < len(trimmed) and trimmed[pos].isdigit():
	pos += 1

	number_str = trimmed[:pos]
	unit_str = trimmed[pos:].strip()

	if not number_str:
	raise ValueError("text does not start with a number")

	try:
	value = int(number_str)
	except ValueError:
	raise ValueError(
	f"The value '{number_str}' cannot be re represented as 64bit number (numeric overflow)."
	)

	if not unit_str:
	result_ms = value
	elif unit_str in ('ns', 'nano', 'nanosecond', 'nanoseconds'):
	result_ms = value / 1_000_000
	elif unit_str in ('µs', 'micro', 'microsecond', 'microseconds'):
	result_ms = value / 1_000
	elif unit_str in ('ms', 'milli', 'millisecond', 'milliseconds'):
	result_ms = value
	elif unit_str in ('s', 'sec', 'second', 'seconds'):
	result_ms = value * 1000
	elif unit_str in ('m', 'min', 'minute', 'minutes'):
	result_ms = value * 60 * 1000
	elif unit_str in ('h', 'hour', 'hours'):
	result_ms = value * 60 * 60 * 1000
	elif unit_str in ('d', 'day', 'days'):
	result_ms = value * 24 * 60 * 60 * 1000
	else:
	supported_units = (
	'DAYS: (d \| day \| days), '
	'HOURS: (h \| hour \| hours), '
	'MINUTES: (m \| min \| minute \| minutes), '
	'SECONDS: (s \| sec \| second \| seconds), '
	'MILLISECONDS: (ms \| milli \| millisecond \| milliseconds), '
	'MICROSECONDS: (µs \| micro \| microsecond \| microseconds), '
	'NANOSECONDS: (ns \| nano \| nanosecond \| nanoseconds)'
	)
	raise ValueError(
	f"Time interval unit label '{unit_str}' does not match any of the recognized units: "
	f"{supported_units}"
	)

	result_ms_int = int(round(result_ms))

	if result_ms_int < 0:
	raise ValueError(f"Duration cannot be negative: {text}")

	return result_ms_int


	# Unit label -> nanoseconds. Mirrors the aliases accepted by ``parse_duration``
	# (and Java ``TimeUtils.parseDuration``); a bare number with no unit is treated
	# as milliseconds, matching ``parse_duration``.
	_UNIT_TO_NANOS = {
	'ns': 1, 'nano': 1, 'nanosecond': 1, 'nanoseconds': 1,
	'µs': 1_000, 'micro': 1_000, 'microsecond': 1_000, 'microseconds': 1_000,
	'ms': 1_000_000, 'milli': 1_000_000, 'millisecond': 1_000_000, 'milliseconds': 1_000_000,
	's': 1_000_000_000, 'sec': 1_000_000_000, 'second': 1_000_000_000, 'seconds': 1_000_000_000,
	'm': 60_000_000_000, 'min': 60_000_000_000, 'minute': 60_000_000_000, 'minutes': 60_000_000_000,
	'h': 3_600_000_000_000, 'hour': 3_600_000_000_000, 'hours': 3_600_000_000_000,
	'd': 86_400_000_000_000, 'day': 86_400_000_000_000, 'days': 86_400_000_000_000,
	}


	def parse_duration_nanos(text: str) -> int:
	"""Parse a duration string to an integer nanosecond count.

	This is the full-precision integer counterpart of :func:`parse_duration`
	(which returns rounded milliseconds): it accepts the same unit aliases and
	the same "bare number means milliseconds" convention, but keeps sub-millisecond
	units exactly (``"1ns"`` -> ``1``, ``"500micro"`` -> ``500_000``) instead of
	rounding them to zero. Use it where sub-millisecond precision must not be
	silently dropped. ``parse_duration`` is intentionally left untouched so its
	millisecond contract (relied on by option parsing) does not change.
	"""
	if text is None:
	raise ValueError("text cannot be None")

	trimmed = text.strip().lower()
	if not trimmed:
	raise ValueError("argument is an empty- or whitespace-only string")

	pos = 0
	while pos < len(trimmed) and trimmed[pos].isdigit():
	pos += 1

	number_str = trimmed[:pos]
	unit_str = trimmed[pos:].strip()

	if not number_str:
	raise ValueError("text does not start with a number")

	try:
	value = int(number_str)
	except ValueError:
	raise ValueError(
	f"The value '{number_str}' cannot be re represented as 64bit number (numeric overflow)."
	)

	if not unit_str:
	nanos_per_unit = 1_000_000 # bare number is milliseconds
	elif unit_str in _UNIT_TO_NANOS:
	nanos_per_unit = _UNIT_TO_NANOS[unit_str]
	else:
	supported_units = (
	'DAYS: (d \| day \| days), '
	'HOURS: (h \| hour \| hours), '
	'MINUTES: (m \| min \| minute \| minutes), '
	'SECONDS: (s \| sec \| second \| seconds), '
	'MILLISECONDS: (ms \| milli \| millisecond \| milliseconds), '
	'MICROSECONDS: (µs \| micro \| microsecond \| microseconds), '
	'NANOSECONDS: (ns \| nano \| nanosecond \| nanoseconds)'
	)
	raise ValueError(
	f"Time interval unit label '{unit_str}' does not match any of the recognized units: "
	f"{supported_units}"
	)

	return value * nanos_per_unit


	# ---------------------------------------------------------------------------
	# Codecs for Java temporal types as serialized by Jackson's JavaTimeModule.
	#
	# These mirror the on-disk JSON shapes that Paimon's Java side reads/writes so
	# that a tag file written by pypaimon is interoperable with Java and vice versa
	# (see ``org.apache.paimon.tag.Tag`` and ``TagTest``):
	# - ``LocalDateTime`` -> JSON array ``[year, month, day, hour, minute,
	# second, nanoOfSecond]``
	# - ``Duration`` -> JSON number of (fractional) seconds, e.g. ``86400.0``
	#
	# Resolution note: Python's ``datetime`` / ``timedelta`` are microsecond-based,
	# so a Java create-time or duration finer than a microsecond is truncated to
	# microseconds on read. Tag granularity is coarse, so this is not a concern in
	# practice.
	# ---------------------------------------------------------------------------

	_NANOS_PER_MICRO = 1000


	def local_datetime_to_json_array(dt: datetime) -> List[int]:
	"""Encode a naive ``datetime`` as Java ``LocalDateTime`` array form.

	Matches Jackson's ``LocalDateTimeSerializer`` byte-for-byte: it always emits
	``[year, month, day, hour, minute]``, appends ``second`` only when second or
	nanoOfSecond is non-zero, and appends ``nanoOfSecond`` only when it is
	non-zero -- i.e. trailing zero components are omitted. ``json_array_to_local_datetime``
	pads them back, so a shorter array round-trips. Python ``datetime`` only has
	microsecond resolution, so the emitted nanoOfSecond is ``microsecond * 1000``
	(never finer than microseconds).
	"""
	nano = dt.microsecond * _NANOS_PER_MICRO
	arr = [dt.year, dt.month, dt.day, dt.hour, dt.minute]
	if dt.second != 0 or nano != 0:
	arr.append(dt.second)
	if nano != 0:
	arr.append(nano)
	return arr


	def json_array_to_local_datetime(arr: List[int]) -> datetime:
	"""Decode a Java ``LocalDateTime`` array form into a naive ``datetime``.

	Jackson omits trailing zero components, so the array may be shorter than 7;
	missing components default to 0. A Java nanoOfSecond finer than a
	microsecond is truncated to microseconds (Python's resolution limit).
	"""
	padded = list(arr) + [0] * (7 - len(arr))
	year, month, day, hour, minute, second, nano = padded[:7]
	return datetime(
	year, month, day, hour, minute, second, nano // _NANOS_PER_MICRO
	)


	def duration_to_json_seconds(td: timedelta):
	"""Encode a ``timedelta`` as Java ``Duration`` decimal-seconds number."""
	return td.total_seconds()


	def json_seconds_to_duration(seconds) -> timedelta:
	"""Decode a Java ``Duration`` decimal-seconds number into a ``timedelta``."""
	return timedelta(seconds=seconds)


	def duration_to_iso8601(td: timedelta) -> str:
	"""Render a non-negative ``timedelta`` like ``java.time.Duration.toString()``.

	Examples: 1 day -> ``PT24H`` (Duration has no day unit), 30 min ->
	``PT30M``, 5 s -> ``PT5S``, zero -> ``PT0S``. Matches what Paimon's Java
	``$tags`` system table surfaces for ``time_retained``. Retentions only ever
	come from ``parse_duration``, which rejects negatives, so only the
	non-negative form is supported.
	"""
	total_micros = (
	td.days * 86_400_000_000 + td.seconds * 1_000_000 + td.microseconds
	)
	if total_micros <= 0:
	return "PT0S"

	total_seconds, micros = divmod(total_micros, 1_000_000)
	hours, rem = divmod(total_seconds, 3600)
	minutes, secs = divmod(rem, 60)

	buf = "PT"
	if hours != 0:
	buf += "{}H".format(hours)
	if minutes != 0:
	buf += "{}M".format(minutes)
	if secs != 0 or micros != 0 or buf == "PT":
	frac = ""
	if micros != 0:
	frac = "." + ("%06d" % micros).rstrip("0")
	buf += "{}{}S".format(secs, frac)
	return buf


	def local_datetime_to_millis(dt: datetime) -> int:
	"""Convert a naive ``LocalDateTime`` to epoch millis, treating it as UTC.

	This is the zone-less / wall-clock conversion: it mirrors Java
	``Timestamp.fromLocalDateTime`` and feeds the ``$tags`` system table's
	``create_time`` (a ``TIMESTAMP`` column, which is itself zone-less), so the
	result does NOT depend on the host's time zone. For the REST
	``GetTagResponse`` epoch-millis ``Long`` use
	:func:`local_datetime_to_system_zone_millis` instead. Uses integer math with
	floored sub-millisecond truncation (matching Java ``Instant.toEpochMilli``),
	so it is exact and correct for pre-epoch instants too.
	"""
	return calendar.timegm(dt.timetuple()) * 1000 + dt.microsecond // 1000


	def local_datetime_to_system_zone_millis(dt: datetime) -> int:
	"""Convert a naive ``LocalDateTime`` to epoch millis in the host's default
	time zone.

	Mirrors Java ``LocalDateTime.atZone(ZoneId.systemDefault()).toInstant()
	.toEpochMilli()`` -- the conversion the REST ``GetTagResponse`` applies to
	``tagCreateTime`` (see ``RESTFileSystemCatalog#getTag``). Unlike
	:func:`local_datetime_to_millis` (zone-less / UTC, used by the ``$tags``
	Timestamp column), the result depends on the host's local time zone, so the
	two intentionally differ on non-UTC hosts -- exactly as the two Java paths do.
	"""
	# A naive datetime.timestamp() interprets dt in the system local zone (via
	# mktime). Take whole seconds first to avoid float rounding, then floor the
	# sub-second part to millis to match Instant.toEpochMilli().
	epoch_seconds = int(dt.replace(microsecond=0).timestamp())
	return epoch_seconds * 1000 + dt.microsecond // 1000