release 0.4.5 (#98)

* release 0.4.5

* fix native execution engine datetime issue

* add release notes

README.md

@@ -124,6 +124,13 @@ pre-commit install
 
 ## Update History
 
+### 0.4.5
+
+* [Extension validation](https://github.com/fugue-project/fugue/issues/81) interface and interfaceless syntax
+* Passing dataframes cross workflow ([yield](https://github.com/fugue-project/fugue/pull/94))
+* [OUT TRANSFORM](https://github.com/fugue-project/fugue/issues/82) to transform and finish a branch of execution
+* Fixed a PandasDataFrame datetime [issue](https://github.com/fugue-project/triad/issues/59) that only happened in transformer interface approach
+
 ### 0.4.3
 
 * Unified checkpoints and persist

fugue/execution/native_execution_engine.py

@@ -116,6 +116,11 @@ def map(
  df = to_local_df(df)
  cursor.set(df.peek_array(), 0, 0)
  output_df = map_func(cursor, df)
+ if (
+ isinstance(output_df, PandasDataFrame)
+ and output_df.schema != output_schema
+ ):
+ output_df = PandasDataFrame(output_df.native, output_schema)
  assert_or_throw(
  output_df.schema == output_schema,
  f"map output {output_df.schema} mismatches given {output_schema}",

fugue_test/builtin_suite.py

@@ -863,13 +863,30 @@ def test_col_ops(self):
  a[["x"]].rename(x="xx").assert_eq(ArrayDataFrame([[1], [2]], "xx:long"))
 
  def test_datetime_in_workflow(self):
+ # schema: a:date,b:datetime
+ def t1(df: pd.DataFrame) -> pd.DataFrame:
+ df["b"] = "2020-01-02"
+ df["b"] = pd.to_datetime(df["b"])
+ return df
+
+ class T2(Transformer):
+ def get_output_schema(self, df):
+ return df.schema
+
+ def transform(self, df):
+ # test for issue https://github.com/fugue-project/fugue/issues/92
+ return PandasDataFrame(df.as_pandas())
+
  with self.dag() as dag:
- a = dag.df([["2020-01-01"]], "a:date").transform(transform_datetime_df)
+ a = dag.df([["2020-01-01"]], "a:date").transform(t1)
  b = dag.df(
  [[datetime.date(2020, 1, 1), datetime.datetime(2020, 1, 2)]],
  "a:date,b:datetime",
  )
  b.assert_eq(a)
+ c = dag.df([["2020-01-01", "2020-01-01 00:00:00"]], "a:date,b:datetime")
+ c.transform(T2).assert_eq(c)
+ c.partition(by=["a"]).transform(T2).assert_eq(c)
 
  @pytest.fixture(autouse=True)
  def init_tmpdir(self, tmpdir):
@@ -1189,10 +1206,3 @@ def mock_co_tf4_ex(df1: List[Dict[str, Any]], p=1) -> List[List[Any]]:
  if k == 2:
  raise NotImplementedError
  return [[df1[0]["a"], len(df1), p]]
-
-
-# schema: a:date,b:datetime
-def transform_datetime_df(df: pd.DataFrame) -> pd.DataFrame:
- df["b"] = "2020-01-02"
- df["b"] = pd.to_datetime(df["b"])
- return df

fugue_version/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.3"
+__version__ = "0.4.5"

setup.py

@@ -18,7 +18,7 @@
  keywords="distributed spark dask sql dsl domain specific language",
  url="http://github.com/fugue-project/fugue",
  install_requires=[
- "triad>=0.4.3",
+ "triad>=0.4.5",
  "adagio>=0.2.2",
  "qpd>=0.2.4",
  "sqlalchemy",