Convert data formats before executing each op in LocalExecutor (#280)

* Refactor `LocalExecutor` into more discrete steps that can be overridden

* Adjust how empty transformables are detected

* Fix bug in determination of additional root columns to merge

* Make it possible to pass a merge function in `LocalExecutor`

* Avoid creating nodes when adding empty lists of column names

* Minor cleanup of how additional root columns are merged

* Move data format conversion between DAG nodes into `LocalExecutor`

* Migrate executor test with `DataFrameLike` to `TensorTable`

* Remove `DictArray`

* Add `column_type` method to `TensorTable`

* Adjust CPU-only masking for new enum values

Author: karlhigley

merlin/dag/__init__.py

@@ -15,8 +15,7 @@
 #
 
 # flake8: noqa
-from merlin.dag.base_operator import BaseOperator, Supports
-from merlin.dag.dictarray import DictArray
+from merlin.dag.base_operator import BaseOperator, DataFormats, Supports
 from merlin.dag.graph import Graph
 from merlin.dag.node import Node, iter_nodes, postorder_iter_nodes, preorder_iter_nodes
 from merlin.dag.selector import ColumnSelector

merlin/dag/base_operator.py

@@ -25,7 +25,11 @@
 
 
 class Supports(Flag):
-    """Indicates what type of data representation this operator supports for transformations"""
+    """
+    Indicates what type of data representation this operator supports for transformations
+
+    (Deprecated)
+    """
 
     # cudf dataframe
     CPU_DATAFRAME = auto()
@@ -37,6 +41,19 @@ class Supports(Flag):
     GPU_DICT_ARRAY = auto()
 
 
+class DataFormats(Flag):
+    CUDF_DATAFRAME = auto()
+    PANDAS_DATAFRAME = auto()
+
+    NUMPY_TENSOR_TABLE = auto()
+    CUPY_TENSOR_TABLE = auto()
+    TF_TENSOR_TABLE = auto()
+    TORCH_TENSOR_TABLE = auto()
+
+    NUMPY_DICT_ARRAY = auto()
+    CUPY_DICT_ARRAY = auto()
+
+
 class BaseOperator:
     """
     Base class for all operator classes.
@@ -355,6 +372,15 @@ def supports(self) -> Supports:
         """Returns what kind of data representation this operator supports"""
         return Supports.CPU_DATAFRAME | Supports.GPU_DATAFRAME
 
+    @property
+    def supported_formats(self) -> DataFormats:
+        return (
+            DataFormats.PANDAS_DATAFRAME
+            | DataFormats.CUDF_DATAFRAME
+            | DataFormats.NUMPY_TENSOR_TABLE
+            | DataFormats.CUPY_TENSOR_TABLE
+        )
+
     def _get_columns(self, df, selector):
         if isinstance(df, dict):
             return {col_name: df[col_name] for col_name in selector.names}