addressing comments (#2210)

python/ray/dataframe/dataframe.py

@@ -500,7 +500,7 @@ class DataFrame(object):
             True if the DataFrame is empty.
             False otherwise.
         """
-        return self._row_metadata._empty
+        return len(self.columns) == 0 or len(self.index) == 0
 
     @property
     def values(self):

python/ray/dataframe/index_metadata.py

@@ -5,8 +5,7 @@ import ray
 from .utils import (
     _build_row_lengths,
     _build_col_widths,
-    _build_coord_df,
-    _check_empty)
+    _build_coord_df)
 
 from pandas.core.indexing import convert_to_index_sliceable
 
@@ -49,9 +48,6 @@ class _IndexMetadata(object):
             else:
                 lengths_oid = _build_col_widths.remote(dfs)
             coord_df_oid = _build_coord_df.remote(lengths_oid, index)
-            self._empty = _check_empty.remote(dfs)
-        else:
-            self._empty = True
 
         self._lengths = lengths_oid
         self._coord_df = coord_df_oid
@@ -161,16 +157,6 @@ class _IndexMetadata(object):
     # cache to accept ObjectIDs and ray.get them when needed.
     _index_cache = property(_get_index_cache, _set_index_cache)
 
-    def _get_empty(self):
-        if isinstance(self._empty_cache, ray.ObjectID):
-            self._empty_cache = ray.get(self._empty_cache)
-        return self._empty_cache
-
-    def _set_empty(self, empty):
-        self._empty_cache = empty
-
-    _empty = property(_get_empty, _set_empty)
-
     def coords_of(self, key):
         """Returns the coordinates (partition, index_within_partition) of the
         provided key in the index. Can be called on its own or implicitly

python/ray/dataframe/utils.py

@@ -155,11 +155,6 @@ def _get_widths(df):
         return 0
 
 
-def _get_empty(df):
-    """Return True if the DataFrame is empty"""
-    return df.empty
-
-
 def _partition_pandas_dataframe(df, num_partitions=None, row_chunksize=None):
     """Partitions a Pandas DataFrame object.
     Args:
@@ -356,12 +351,6 @@ def _build_coord_df(lengths, index):
     return pandas.DataFrame(coords, index=index, columns=col_names)
 
 
-@ray.remote
-def _check_empty(dfs):
-    """Check if all partitions are empty"""
-    return all(ray.get([_deploy_func.remote(_get_empty, d) for d in dfs]))
-
-
 def _create_block_partitions(partitions, axis=0, length=None):
 
     if length is not None and length != 0 and get_npartitions() > length: