FIX-#4014: Apply deferred metadata before data egress (#4025)

Signed-off-by: Devin Petersohn <devin.petersohn@gmail.com>

modin/core/execution/ray/implementations/pandas_on_ray/io/io.py

@@ -111,6 +111,8 @@ def func(df):
             df.to_sql(**kwargs)
             return pandas.DataFrame()
 
+        # Ensure that the metadata is syncrhonized
+        qc._modin_frame._propagate_index_objs(axis=None)
         result = qc._modin_frame.apply_full_axis(1, func, new_index=[], new_columns=[])
         # FIXME: we should be waiting for completion less expensievely, maybe use _modin_frame.materialize()?
         result.to_pandas()  # blocking operation
@@ -225,6 +227,8 @@ def func(df, **kw):
 
         # signaling that the partition with id==0 can be written to the file
         ray.get(signals.send.remote(0))
+        # Ensure that the metadata is syncrhonized
+        qc._modin_frame._propagate_index_objs(axis=None)
         result = qc._modin_frame._partition_mgr_cls.map_axis_partitions(
             axis=1,
             partitions=qc._modin_frame._partitions,

modin/pandas/test/test_io.py

@@ -1166,6 +1166,26 @@ def test_read_csv_skiprows_corner_cases(self, skiprows):
             skiprows=skiprows,
         )
 
+    def test_to_csv_with_index(self):
+        cols = 100
+        arows = 20000
+        keyrange = 100
+        values = np.vstack(
+            [
+                np.random.choice(keyrange, size=(arows)),
+                np.random.normal(size=(cols, arows)),
+            ]
+        ).transpose()
+        modin_df = pd.DataFrame(
+            values,
+            columns=["key"] + ["avalue" + str(i) for i in range(1, 1 + cols)],
+        ).set_index("key")
+        pandas_df = pandas.DataFrame(
+            values,
+            columns=["key"] + ["avalue" + str(i) for i in range(1, 1 + cols)],
+        ).set_index("key")
+        eval_to_file(modin_df, pandas_df, "to_csv", "csv")
+
 
 class TestTable:
     def test_read_table(self, make_csv_file):