-
Notifications
You must be signed in to change notification settings - Fork 71
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Feature/datetime functions #91
Merged
nils-braun
merged 6 commits into
dask-contrib:main
from
Yuhuishishishi:feature/datetime-functions
Dec 7, 2020
Merged
Changes from all commits
Commits
Show all changes
6 commits
Select commit
Hold shift + click to select a range
47ecbdb
Implement the last_day function
Yuhuishishishi 8d549d0
Modify interval literals for timestampadd
Yuhuishishishi 6289018
Ceil and floor for datetime
Yuhuishishishi 8e0a268
Clean up docs and comments
Yuhuishishishi 467b28a
Code formatting + common util functions
Yuhuishishishi a608483
Add tests for exception handling
Yuhuishishishi File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -14,7 +14,7 @@ | |
|
||
from dask_sql.physical.rex import RexConverter | ||
from dask_sql.physical.rex.base import BaseRexPlugin | ||
from dask_sql.utils import LoggableDataFrame, is_frame | ||
from dask_sql.utils import LoggableDataFrame, is_frame, is_datetime, convert_to_datetime | ||
from dask_sql.datacontainer import DataContainer | ||
|
||
logger = logging.getLogger(__name__) | ||
|
@@ -37,25 +37,34 @@ def of(self, op: "Operation") -> "Operation": | |
return Operation(lambda x: self(op(x))) | ||
|
||
|
||
class TensorScalarOperation(Operation): | ||
class PredicteBasedOperation(Operation): | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Nice one! |
||
""" | ||
Helper operation to call a function on the input, | ||
depending if the first is a dataframe or not | ||
depending if the first arg evaluates, given a predicate function, to true or false | ||
""" | ||
|
||
def __init__(self, tensor_f: Callable, scalar_f: Callable = None): | ||
"""Init with the given operation""" | ||
def __init__(self, predicte: Callable, true_route: Callable, false_route: Callable): | ||
super().__init__(self.apply) | ||
|
||
self.tensor_f = tensor_f | ||
self.scalar_f = scalar_f or tensor_f | ||
self.predicte = predicte | ||
self.true_route = true_route | ||
self.false_route = false_route | ||
|
||
def apply(self, *operands): | ||
"""Call the stored functions""" | ||
if is_frame(operands[0]): | ||
return self.tensor_f(*operands) | ||
if self.predicte(operands[0]): | ||
return self.true_route(*operands) | ||
|
||
return self.scalar_f(*operands) | ||
return self.false_route(*operands) | ||
|
||
|
||
class TensorScalarOperation(PredicteBasedOperation): | ||
""" | ||
Helper operation to call a function on the input, | ||
depending if the first is a dataframe or not | ||
""" | ||
|
||
def __init__(self, tensor_f: Callable, scalar_f: Callable = None): | ||
"""Init with the given operation""" | ||
super().__init__(is_frame, tensor_f, scalar_f) | ||
|
||
|
||
class ReduceOperation(Operation): | ||
|
@@ -377,10 +386,7 @@ def __init__(self): | |
|
||
def extract(self, what, df: SeriesOrScalar): | ||
input_df = df | ||
if is_frame(df): | ||
df = df.dt | ||
else: | ||
df = pd.to_datetime(df) | ||
df = convert_to_datetime(df) | ||
|
||
if what == "CENTURY": | ||
return da.trunc(df.year / 100) | ||
|
@@ -416,6 +422,48 @@ def extract(self, what, df: SeriesOrScalar): | |
raise NotImplementedError(f"Extraction of {what} is not (yet) implemented.") | ||
|
||
|
||
class CeilFloorOperation(PredicteBasedOperation): | ||
""" | ||
Apply ceil/floor operations on a series depending on its dtype (datetime like vs normal) | ||
""" | ||
|
||
def __init__(self, round_method: str): | ||
assert round_method in { | ||
"ceil", | ||
"floor", | ||
}, f"Round method can only be either ceil or floor" | ||
|
||
super().__init__( | ||
is_datetime, # if the series is dt type | ||
self._round_datetime, | ||
getattr(da, round_method), | ||
) | ||
|
||
self.round_method = round_method | ||
|
||
def _round_datetime(self, *operands): | ||
df, unit = operands | ||
|
||
df = convert_to_datetime(df) | ||
|
||
unit_map = { | ||
"DAY": "D", | ||
"HOUR": "H", | ||
"MINUTE": "T", | ||
"SECOND": "S", | ||
"MICROSECOND": "U", | ||
"MILLISECOND": "L", | ||
} | ||
|
||
try: | ||
freq = unit_map[unit.upper()] | ||
return getattr(df, self.round_method)(freq) | ||
except KeyError: | ||
raise NotImplementedError( | ||
f"{self.round_method} TO {unit} is not (yet) implemented." | ||
) | ||
|
||
|
||
class RexCallPlugin(BaseRexPlugin): | ||
""" | ||
RexCall is used for expressions, which calculate something. | ||
|
@@ -468,12 +516,12 @@ class RexCallPlugin(BaseRexPlugin): | |
"atan": Operation(da.arctan), | ||
"atan2": Operation(da.arctan2), | ||
"cbrt": Operation(da.cbrt), | ||
"ceil": Operation(da.ceil), | ||
"ceil": CeilFloorOperation("ceil"), | ||
"cos": Operation(da.cos), | ||
"cot": Operation(lambda x: 1 / da.tan(x)), | ||
"degrees": Operation(da.degrees), | ||
"exp": Operation(da.exp), | ||
"floor": Operation(da.floor), | ||
"floor": CeilFloorOperation("floor"), | ||
"log10": Operation(da.log10), | ||
"ln": Operation(da.log), | ||
# "mod": Operation(da.mod), # needs cast | ||
|
@@ -501,6 +549,10 @@ class RexCallPlugin(BaseRexPlugin): | |
"current_time": Operation(lambda *args: pd.Timestamp.now()), | ||
"current_date": Operation(lambda *args: pd.Timestamp.now()), | ||
"current_timestamp": Operation(lambda *args: pd.Timestamp.now()), | ||
"last_day": TensorScalarOperation( | ||
lambda x: x + pd.tseries.offsets.MonthEnd(1), | ||
lambda x: convert_to_datetime(x) + pd.tseries.offsets.MonthEnd(1), | ||
), | ||
} | ||
|
||
def convert( | ||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What is the benefit of your class (
pd.tseries.offsets.DateOffset
) to thedatetime
timedelta? Just for me to understand - I am very fine with changing itThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
With
pd.tseries.offsets
, my understanding is that it's more convenient to specify timedelta with data-dependent length. E.g., when incrementing the current timestamp by "1 month", with standard timedelta, we need to compute if to add 30/31/28/29 days depending on the month/year of the current timestamp.With
pd.tseries.offsets
, we can use `pd.tseries.offsets.DateOffset(months=1), and pandas shall take care of the day conversion under the hood.Similar thing happens to "year" increment as well, due to leap years.
For
last_day
implementation, it is convenient to usepd.tseries.offsets.MonthEnd
, which shifts the time to the last day of the month.For more generic cases (such as + hours/minutes/seconds), I think the standard timedelta and
pd.tseries.offsets
will all do the tricks.