We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fill Missing Op option to create a new binary column indicating the value was replaced.
Fill Missing Op should have the option to create a new binary column indicating whether the column was filled or not for continuous variables.

This is a common feature used when dealing with missing values of categoricals.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nvtabular/ops/fill.py`
Content:
```
1 #
2 # Copyright (c) 2020, NVIDIA CORPORATION.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 #
16 import cudf
17 import dask_cudf
18 from nvtx import annotate
19 
20 from .operator import ColumnNames, Operator
21 from .stat_operator import StatOperator
22 
23 
24 class FillMissing(Operator):
25     """
26     This operation replaces missing values with a constant pre-defined value
27 
28     Example usage::
29 
30         # Use FillMissing to define a workflow for continuous columns and specify the fill value
31         # Default is 0
32         cont_features = ['cont1', 'cont2', 'cont3'] >> ops.FillMissing() >> ...
33         processor = nvtabular.Workflow(cont_features)
34 
35     Parameters
36     -----------
37     fill_val : float, default 0
38         The constant value to replace missing values with.
39     """
40 
41     def __init__(self, fill_val=0):
42         super().__init__()
43         self.fill_val = fill_val
44 
45     @annotate("FillMissing_op", color="darkgreen", domain="nvt_python")
46     def transform(self, columns, gdf: cudf.DataFrame) -> cudf.DataFrame:
47         return gdf[columns].fillna(self.fill_val)
48 
49     transform.__doc__ = Operator.transform.__doc__
50 
51 
52 class FillMedian(StatOperator):
53     """
54     This operation replaces missing values with the median value for the column.
55 
56     Example usage::
57 
58         # Initialize the workflow
59         proc = nvt.Workflow(
60             cat_names=CATEGORICAL_COLUMNS,
61             cont_names=CONTINUOUS_COLUMNS,
62             label_name=LABEL_COLUMNS
63         )
64 
65         # Add FillMedian to the workflow for continuous columns
66         proc.add_cont_feature(nvt.ops.FillMedian())
67     """
68 
69     def __init__(self):
70         super().__init__()
71         self.medians = {}
72 
73     @annotate("FillMedian_transform", color="darkgreen", domain="nvt_python")
74     def transform(self, columns: ColumnNames, gdf: cudf.DataFrame) -> cudf.DataFrame:
75         if not self.medians:
76             raise RuntimeError("need to call 'fit' before running transform")
77 
78         for col in columns:
79             gdf[col] = gdf[col].fillna(self.medians[col])
80         return gdf
81 
82     @annotate("FillMedian_fit", color="green", domain="nvt_python")
83     def fit(self, columns: ColumnNames, ddf: dask_cudf.DataFrame):
84         # TODO: Use `method="tidigest"` when crick supports device
85         dask_stats = ddf[columns].quantile(q=0.5, method="dask")
86         return dask_stats
87 
88     @annotate("FillMedian_finalize", color="green", domain="nvt_python")
89     def fit_finalize(self, dask_stats):
90         for col in dask_stats.index.values_host:
91             self.medians[col] = float(dask_stats[col])
92 
93     transform.__doc__ = Operator.transform.__doc__
94     fit.__doc__ = StatOperator.fit.__doc__
95     fit_finalize.__doc__ = StatOperator.fit_finalize.__doc__
96 
97     def clear(self):
98         self.medians = {}
99
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/nvtabular/ops/fill.py b/nvtabular/ops/fill.py
--- a/nvtabular/ops/fill.py
+++ b/nvtabular/ops/fill.py
@@ -36,18 +36,34 @@
     -----------
     fill_val : float, default 0
         The constant value to replace missing values with.
+    add_binary_cols : boolean, default False
+        When True, adds binary columns that indicate whether cells in each column were filled
     """
 
-    def __init__(self, fill_val=0):
+    def __init__(self, fill_val=0, add_binary_cols=False):
         super().__init__()
         self.fill_val = fill_val
+        self.add_binary_cols = add_binary_cols
 
     @annotate("FillMissing_op", color="darkgreen", domain="nvt_python")
     def transform(self, columns, gdf: cudf.DataFrame) -> cudf.DataFrame:
-        return gdf[columns].fillna(self.fill_val)
+        if self.add_binary_cols:
+            for col in columns:
+                gdf[f"{col}_filled"] = gdf[col].isna()
+                gdf[col] = gdf[col].fillna(self.fill_val)
+        else:
+            gdf[columns] = gdf[columns].fillna(self.fill_val)
+
+        return gdf
 
     transform.__doc__ = Operator.transform.__doc__
 
+    def output_column_names(self, columns: ColumnNames) -> ColumnNames:
+        output_cols = columns[:]
+        if self.add_binary_cols:
+            output_cols.extend([f"{col}_filled" for col in columns])
+        return output_cols
+
 
 class FillMedian(StatOperator):
     """
@@ -64,10 +80,16 @@
 
         # Add FillMedian to the workflow for continuous columns
         proc.add_cont_feature(nvt.ops.FillMedian())
+
+    Parameters
+    -----------
+    add_binary_cols : boolean, default False
+        When True, adds binary columns that indicate whether cells in each column were filled
     """
 
-    def __init__(self):
+    def __init__(self, add_binary_cols=False):
         super().__init__()
+        self.add_binary_cols = add_binary_cols
         self.medians = {}
 
     @annotate("FillMedian_transform", color="darkgreen", domain="nvt_python")
@@ -76,6 +98,8 @@
             raise RuntimeError("need to call 'fit' before running transform")
 
         for col in columns:
+            if self.add_binary_cols:
+                gdf[f"{col}_filled"] = gdf[col].isna()
             gdf[col] = gdf[col].fillna(self.medians[col])
         return gdf
 
@@ -96,3 +120,9 @@
 
     def clear(self):
         self.medians = {}
+
+    def output_column_names(self, columns: ColumnNames) -> ColumnNames:
+        output_cols = columns[:]
+        if self.add_binary_cols:
+            output_cols.extend([f"{col}_filled" for col in columns])
+        return output_cols

{"golden_diff": "diff --git a/nvtabular/ops/fill.py b/nvtabular/ops/fill.py\n--- a/nvtabular/ops/fill.py\n+++ b/nvtabular/ops/fill.py\n@@ -36,18 +36,34 @@\n     -----------\n     fill_val : float, default 0\n         The constant value to replace missing values with.\n+    add_binary_cols : boolean, default False\n+        When True, adds binary columns that indicate whether cells in each column were filled\n     \"\"\"\n \n-    def __init__(self, fill_val=0):\n+    def __init__(self, fill_val=0, add_binary_cols=False):\n         super().__init__()\n         self.fill_val = fill_val\n+        self.add_binary_cols = add_binary_cols\n \n     @annotate(\"FillMissing_op\", color=\"darkgreen\", domain=\"nvt_python\")\n     def transform(self, columns, gdf: cudf.DataFrame) -> cudf.DataFrame:\n-        return gdf[columns].fillna(self.fill_val)\n+        if self.add_binary_cols:\n+            for col in columns:\n+                gdf[f\"{col}_filled\"] = gdf[col].isna()\n+                gdf[col] = gdf[col].fillna(self.fill_val)\n+        else:\n+            gdf[columns] = gdf[columns].fillna(self.fill_val)\n+\n+        return gdf\n \n     transform.__doc__ = Operator.transform.__doc__\n \n+    def output_column_names(self, columns: ColumnNames) -> ColumnNames:\n+        output_cols = columns[:]\n+        if self.add_binary_cols:\n+            output_cols.extend([f\"{col}_filled\" for col in columns])\n+        return output_cols\n+\n \n class FillMedian(StatOperator):\n     \"\"\"\n@@ -64,10 +80,16 @@\n \n         # Add FillMedian to the workflow for continuous columns\n         proc.add_cont_feature(nvt.ops.FillMedian())\n+\n+    Parameters\n+    -----------\n+    add_binary_cols : boolean, default False\n+        When True, adds binary columns that indicate whether cells in each column were filled\n     \"\"\"\n \n-    def __init__(self):\n+    def __init__(self, add_binary_cols=False):\n         super().__init__()\n+        self.add_binary_cols = add_binary_cols\n         self.medians = {}\n \n     @annotate(\"FillMedian_transform\", color=\"darkgreen\", domain=\"nvt_python\")\n@@ -76,6 +98,8 @@\n             raise RuntimeError(\"need to call 'fit' before running transform\")\n \n         for col in columns:\n+            if self.add_binary_cols:\n+                gdf[f\"{col}_filled\"] = gdf[col].isna()\n             gdf[col] = gdf[col].fillna(self.medians[col])\n         return gdf\n \n@@ -96,3 +120,9 @@\n \n     def clear(self):\n         self.medians = {}\n+\n+    def output_column_names(self, columns: ColumnNames) -> ColumnNames:\n+        output_cols = columns[:]\n+        if self.add_binary_cols:\n+            output_cols.extend([f\"{col}_filled\" for col in columns])\n+        return output_cols\n", "issue": "Fill Missing Op option to create a new binary column indicating the value was replaced.\nFill Missing Op should have the option to create a new binary column indicating whether the column was filled or not for continuous variables.\r\n\r\nThis is a common feature used when dealing with missing values of categoricals.\n", "before_files": [{"content": "#\n# Copyright (c) 2020, NVIDIA CORPORATION.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\nimport cudf\nimport dask_cudf\nfrom nvtx import annotate\n\nfrom .operator import ColumnNames, Operator\nfrom .stat_operator import StatOperator\n\n\nclass FillMissing(Operator):\n    \"\"\"\n    This operation replaces missing values with a constant pre-defined value\n\n    Example usage::\n\n        # Use FillMissing to define a workflow for continuous columns and specify the fill value\n        # Default is 0\n        cont_features = ['cont1', 'cont2', 'cont3'] >> ops.FillMissing() >> ...\n        processor = nvtabular.Workflow(cont_features)\n\n    Parameters\n    -----------\n    fill_val : float, default 0\n        The constant value to replace missing values with.\n    \"\"\"\n\n    def __init__(self, fill_val=0):\n        super().__init__()\n        self.fill_val = fill_val\n\n    @annotate(\"FillMissing_op\", color=\"darkgreen\", domain=\"nvt_python\")\n    def transform(self, columns, gdf: cudf.DataFrame) -> cudf.DataFrame:\n        return gdf[columns].fillna(self.fill_val)\n\n    transform.__doc__ = Operator.transform.__doc__\n\n\nclass FillMedian(StatOperator):\n    \"\"\"\n    This operation replaces missing values with the median value for the column.\n\n    Example usage::\n\n        # Initialize the workflow\n        proc = nvt.Workflow(\n            cat_names=CATEGORICAL_COLUMNS,\n            cont_names=CONTINUOUS_COLUMNS,\n            label_name=LABEL_COLUMNS\n        )\n\n        # Add FillMedian to the workflow for continuous columns\n        proc.add_cont_feature(nvt.ops.FillMedian())\n    \"\"\"\n\n    def __init__(self):\n        super().__init__()\n        self.medians = {}\n\n    @annotate(\"FillMedian_transform\", color=\"darkgreen\", domain=\"nvt_python\")\n    def transform(self, columns: ColumnNames, gdf: cudf.DataFrame) -> cudf.DataFrame:\n        if not self.medians:\n            raise RuntimeError(\"need to call 'fit' before running transform\")\n\n        for col in columns:\n            gdf[col] = gdf[col].fillna(self.medians[col])\n        return gdf\n\n    @annotate(\"FillMedian_fit\", color=\"green\", domain=\"nvt_python\")\n    def fit(self, columns: ColumnNames, ddf: dask_cudf.DataFrame):\n        # TODO: Use `method=\"tidigest\"` when crick supports device\n        dask_stats = ddf[columns].quantile(q=0.5, method=\"dask\")\n        return dask_stats\n\n    @annotate(\"FillMedian_finalize\", color=\"green\", domain=\"nvt_python\")\n    def fit_finalize(self, dask_stats):\n        for col in dask_stats.index.values_host:\n            self.medians[col] = float(dask_stats[col])\n\n    transform.__doc__ = Operator.transform.__doc__\n    fit.__doc__ = StatOperator.fit.__doc__\n    fit_finalize.__doc__ = StatOperator.fit_finalize.__doc__\n\n    def clear(self):\n        self.medians = {}\n", "path": "nvtabular/ops/fill.py"}], "after_files": [{"content": "#\n# Copyright (c) 2020, NVIDIA CORPORATION.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\nimport cudf\nimport dask_cudf\nfrom nvtx import annotate\n\nfrom .operator import ColumnNames, Operator\nfrom .stat_operator import StatOperator\n\n\nclass FillMissing(Operator):\n    \"\"\"\n    This operation replaces missing values with a constant pre-defined value\n\n    Example usage::\n\n        # Use FillMissing to define a workflow for continuous columns and specify the fill value\n        # Default is 0\n        cont_features = ['cont1', 'cont2', 'cont3'] >> ops.FillMissing() >> ...\n        processor = nvtabular.Workflow(cont_features)\n\n    Parameters\n    -----------\n    fill_val : float, default 0\n        The constant value to replace missing values with.\n    add_binary_cols : boolean, default False\n        When True, adds binary columns that indicate whether cells in each column were filled\n    \"\"\"\n\n    def __init__(self, fill_val=0, add_binary_cols=False):\n        super().__init__()\n        self.fill_val = fill_val\n        self.add_binary_cols = add_binary_cols\n\n    @annotate(\"FillMissing_op\", color=\"darkgreen\", domain=\"nvt_python\")\n    def transform(self, columns, gdf: cudf.DataFrame) -> cudf.DataFrame:\n        if self.add_binary_cols:\n            for col in columns:\n                gdf[f\"{col}_filled\"] = gdf[col].isna()\n                gdf[col] = gdf[col].fillna(self.fill_val)\n        else:\n            gdf[columns] = gdf[columns].fillna(self.fill_val)\n\n        return gdf\n\n    transform.__doc__ = Operator.transform.__doc__\n\n    def output_column_names(self, columns: ColumnNames) -> ColumnNames:\n        output_cols = columns[:]\n        if self.add_binary_cols:\n            output_cols.extend([f\"{col}_filled\" for col in columns])\n        return output_cols\n\n\nclass FillMedian(StatOperator):\n    \"\"\"\n    This operation replaces missing values with the median value for the column.\n\n    Example usage::\n\n        # Initialize the workflow\n        proc = nvt.Workflow(\n            cat_names=CATEGORICAL_COLUMNS,\n            cont_names=CONTINUOUS_COLUMNS,\n            label_name=LABEL_COLUMNS\n        )\n\n        # Add FillMedian to the workflow for continuous columns\n        proc.add_cont_feature(nvt.ops.FillMedian())\n\n    Parameters\n    -----------\n    add_binary_cols : boolean, default False\n        When True, adds binary columns that indicate whether cells in each column were filled\n    \"\"\"\n\n    def __init__(self, add_binary_cols=False):\n        super().__init__()\n        self.add_binary_cols = add_binary_cols\n        self.medians = {}\n\n    @annotate(\"FillMedian_transform\", color=\"darkgreen\", domain=\"nvt_python\")\n    def transform(self, columns: ColumnNames, gdf: cudf.DataFrame) -> cudf.DataFrame:\n        if not self.medians:\n            raise RuntimeError(\"need to call 'fit' before running transform\")\n\n        for col in columns:\n            if self.add_binary_cols:\n                gdf[f\"{col}_filled\"] = gdf[col].isna()\n            gdf[col] = gdf[col].fillna(self.medians[col])\n        return gdf\n\n    @annotate(\"FillMedian_fit\", color=\"green\", domain=\"nvt_python\")\n    def fit(self, columns: ColumnNames, ddf: dask_cudf.DataFrame):\n        # TODO: Use `method=\"tidigest\"` when crick supports device\n        dask_stats = ddf[columns].quantile(q=0.5, method=\"dask\")\n        return dask_stats\n\n    @annotate(\"FillMedian_finalize\", color=\"green\", domain=\"nvt_python\")\n    def fit_finalize(self, dask_stats):\n        for col in dask_stats.index.values_host:\n            self.medians[col] = float(dask_stats[col])\n\n    transform.__doc__ = Operator.transform.__doc__\n    fit.__doc__ = StatOperator.fit.__doc__\n    fit_finalize.__doc__ = StatOperator.fit_finalize.__doc__\n\n    def clear(self):\n        self.medians = {}\n\n    def output_column_names(self, columns: ColumnNames) -> ColumnNames:\n        output_cols = columns[:]\n        if self.add_binary_cols:\n            output_cols.extend([f\"{col}_filled\" for col in columns])\n        return output_cols\n", "path": "nvtabular/ops/fill.py"}]}