Revert "Check out a mutable table info to fix a race condition where …

…we potentially attempt to fix a column multiple times per df"

This reverts commit 142ac8a.

morpheus/_lib/src/stages/preprocess_fil.cpp

@@ -121,8 +121,8 @@ PreprocessFILStage::subscribe_fn_t PreprocessFILStage::build_operator()
                                                                        input__0.get_memory(),
                                                                        x->mess_offset),
                                               seq_id_dtype,
-                                                   {x->mess_count, 3},
-                                                   {},
+                                              {x->mess_count, 3},
+                                              {},
                                               0);
 
                 // Build the results
@@ -134,12 +134,8 @@ PreprocessFILStage::subscribe_fn_t PreprocessFILStage::build_operator()
 
                 output.on_next(std::move(next));
             },
-            [&](std::exception_ptr error_ptr) {
-                output.on_error(error_ptr);
-            },
-            [&]() {
-                output.on_completed();
-            }));
+            [&](std::exception_ptr error_ptr) { output.on_error(error_ptr); },
+            [&]() { output.on_completed(); }));
     };
 }
 
@@ -148,43 +144,50 @@ TableInfo PreprocessFILStage::fix_bad_columns(sink_type_t x)
     std::vector<std::string> bad_cols;
 
     {
-        // Get the mutable info for the entire meta object so we only do this once per dataframe
-        auto mutable_info      = x->meta->get_mutable_info();
-        auto df_meta_col_names = mutable_info.get_column_names();
+        // TODO(MDD): Add some sort of lock here to prevent fixing columns after they have been accessed
+        auto df_meta           = x->get_meta(m_fea_cols);
+        auto df_meta_col_names = df_meta.get_column_names();
 
-        for (size_t i = 0; i < mutable_info.num_columns(); ++i)
+        for (size_t i = 0; i < df_meta.num_columns(); ++i)
         {
-            if (mutable_info.get_column(i).type().id() == cudf::type_id::STRING)
+            if (df_meta.get_column(i).type().id() == cudf::type_id::STRING)
             {
                 bad_cols.push_back(df_meta_col_names[i]);
             }
         }
 
         // Exit early if there is nothing to do
-        if (!bad_cols.empty())
+        if (bad_cols.empty())
         {
-            // Need to ensure all string columns have been converted to numbers. This requires running a
-            // regex which is too difficult to do from C++ at this time. So grab the GIL, make the
-            // conversions, and release. This is horribly inefficient, but so is the JSON lines format for
-            // this workflow
-            using namespace pybind11::literals;
-            pybind11::gil_scoped_acquire gil;
+            return df_meta;
+        }
+    }
 
-            // pybind11::object df = x->meta->get_py_table();
-            auto df = mutable_info.checkout_obj();
+    // Need to ensure all string columns have been converted to numbers. This requires running a
+    // regex which is too difficult to do from C++ at this time. So grab the GIL, make the
+    // conversions, and release. This is horribly inefficient, but so is the JSON lines format for
+    // this workflow
+    {
+        // Get the mutable info for the entire meta object so we only do this once per dataframe
+        auto mutable_info = x->meta->get_mutable_info();
 
-            std::string regex = R"((\d+))";
+        using namespace pybind11::literals;
+        pybind11::gil_scoped_acquire gil;
 
-            for (auto c : bad_cols)
-            {
-                df[pybind11::str(c)] = df[pybind11::str(c)]
-                                           .attr("str")
-                                           .attr("extract")(pybind11::str(regex), "expand"_a = true)
-                                           .attr("astype")(pybind11::str("float32"));
-            }
+        // pybind11::object df = x->meta->get_py_table();
+        auto df = mutable_info.checkout_obj();
+
+        std::string regex = R"((\d+))";
 
-            mutable_info.return_obj(std::move(df));
+        for (auto c : bad_cols)
+        {
+            df[pybind11::str(c)] = df[pybind11::str(c)]
+                                       .attr("str")
+                                       .attr("extract")(pybind11::str(regex), "expand"_a = true)
+                                       .attr("astype")(pybind11::str("float32"));
         }
+
+        mutable_info.return_obj(std::move(df));
     }
 
     // Now re-get the meta