prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Failing to open arrow stream with categoricals
  Vaex fails to open arrow streams that contain categorical columns. It would be great if this was working as categoricals have given a great performance in many of my applications.
  
  ```import pandas as pd
  import pyarrow as pa
  import numpy as np
  import vaex
  df = pd.DataFrame(
      {
          'col1': pd.Categorical.from_codes(np.full(1, 1), categories=['ABC', 'DEF'])
      }
  )
  table = pa.Table.from_pandas(df)
  
  with pa.OSFile('test2.arrow', 'wb') as sink:
      with pa.RecordBatchStreamWriter(sink, table.schema) as writer:
          writer.write_table(table)
  
  with pa.OSFile('test2.arrow', 'rb') as source:
      df = pa.ipc.open_stream(source).read_pandas()
  
  df = vaex.open('test2.arrow')
  ```
  
  Output:
  ```
  ERROR:MainThread:vaex:error opening 'test2.arrow'
  ---------------------------------------------------------------------------
  NotImplementedError                       Traceback (most recent call last)
   in 
       17     df = pa.ipc.open_stream(source).read_pandas()
       18 
  ---> 19 df = vaex.open('test2.arrow')
  
  Z:\Systemdateien\Miniconda3\envs\finance\lib\site-packages\vaex\__init__.py in open(path, convert, shuffle, copy_index, *args, **kwargs)
      189                         ds = from_csv(path, copy_index=copy_index, **kwargs)
      190                     else:
  --> 191                         ds = vaex.file.open(path, *args, **kwargs)
      192                     if convert and ds:
      193                         ds.export_hdf5(filename_hdf5, shuffle=shuffle)
  
  Z:\Systemdateien\Miniconda3\envs\finance\lib\site-packages\vaex\file\__init__.py in open(path, *args, **kwargs)
       28     for opener in opener_classes:
       29         if opener.can_open(path, *args, **kwargs):
  ---> 30             return opener.open(path, *args, **kwargs)
       31     if hdf5:
       32         openers.extend(hdf5.dataset.dataset_type_map.items())
  
  Z:\Systemdateien\Miniconda3\envs\finance\lib\site-packages\vaex_arrow\opener.py in open(path, *args, **kwargs)
        9     def open(path, *args, **kwargs):
       10         from .dataset import DatasetArrow
  ---> 11         return DatasetArrow(path, *args, **kwargs)
       12 
       13 class ParquetOpener:
  
  Z:\Systemdateien\Miniconda3\envs\finance\lib\site-packages\vaex_arrow\dataset.py in __init__(self, filename, table, write)
       18         self._write = write
       19         if table is None:
  ---> 20             self._load()
       21         else:
       22             self._load_table(table)
  
  Z:\Systemdateien\Miniconda3\envs\finance\lib\site-packages\vaex_arrow\dataset.py in _load(self)
       26         reader = pa.ipc.open_stream(source)
       27         table = pa.Table.from_batches([b for b in reader])
  ---> 28         self._load_table(table)
       29 
       30     def _load_table(self, table):
  
  Z:\Systemdateien\Miniconda3\envs\finance\lib\site-packages\vaex_arrow\dataset.py in _load_table(self, table)
       35             # TODO: keep the arrow columns, and support and test chunks
       36             arrow_array = col.data.chunks[0]
  ---> 37             column = column_from_arrow_array(arrow_array)
       38 
       39             self.columns[name] = column
  
  Z:\Systemdateien\Miniconda3\envs\finance\lib\site-packages\vaex_arrow\convert.py in column_from_arrow_array(arrow_array)
       28     buffers = arrow_array.buffers()
       29     if len(buffers) == 2:
  ---> 30         return numpy_array_from_arrow_array(arrow_array)
       31     elif len(buffers) == 3 and  isinstance(arrow_array.type, type(pyarrow.string())):
       32         bitmap_buffer, offsets, string_bytes = arrow_array.buffers()
  
  Z:\Systemdateien\Miniconda3\envs\finance\lib\site-packages\vaex_arrow\convert.py in numpy_array_from_arrow_array(arrow_array)
       62         array = np.frombuffer(buffers[-1], dtype, len(arrow_array))# TODO: deal with offset ? [arrow_array.offset:arrow_array.offset + len(arrow_array)]
       63     else:
  ---> 64         dtype = arrow_array.type.to_pandas_dtype()
       65     if np.bool_ == dtype:
       66         # TODO: this will also be a copy, we probably want to support bitmasks as well
  
  Z:\Systemdateien\Miniconda3\envs\finance\lib\site-packages\pyarrow\types.pxi in pyarrow.lib.DataType.to_pandas_dtype()
  
  NotImplementedError: dictionary
  ```
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `packages/vaex-arrow/vaex_arrow/dataset.py`
  Content:
  ```
  1 __author__ = 'maartenbreddels'
  2 import logging
  3 
  4 import pyarrow as pa
  5 import pyarrow.parquet as pq
  6 
  7 import vaex.dataset
  8 import vaex.file.other
  9 from .convert import column_from_arrow_array
  10 logger = logging.getLogger("vaex_arrow")
  11 
  12 
  13 class DatasetArrow(vaex.dataset.DatasetLocal):
  14     """Implements storage using arrow"""
  15 
  16     def __init__(self, filename=None, table=None, write=False):
  17         super(DatasetArrow, self).__init__(name=filename, path=filename, column_names=[])
  18         self._write = write
  19         if table is None:
  20             self._load()
  21         else:
  22             self._load_table(table)
  23 
  24     def _load(self):
  25         source = pa.memory_map(self.path)
  26         reader = pa.ipc.open_stream(source)
  27         table = pa.Table.from_batches([b for b in reader])
  28         self._load_table(table)
  29     
  30     def _load_table(self, table):
  31         self._length_unfiltered =  self._length_original = table.num_rows
  32         self._index_end =  self._length_original = table.num_rows
  33         for col in table.columns:
  34             name = col.name
  35             # TODO: keep the arrow columns, and support and test chunks
  36             arrow_array = col.data.chunks[0]
  37             column = column_from_arrow_array(arrow_array)
  38 
  39             self.columns[name] = column
  40             self.column_names.append(name)
  41             self._save_assign_expression(name, vaex.expression.Expression(self, name))
  42 
  43 
  44     @classmethod
  45     def can_open(cls, path, *args, **kwargs):
  46         return path.rpartition('.')[2] == 'arrow'
  47 
  48     @classmethod
  49     def get_options(cls, path):
  50         return []
  51 
  52     @classmethod
  53     def option_to_args(cls, option):
  54         return []
  55 
  56 class DatasetParquet(DatasetArrow):
  57     def _load(self):
  58         # might not be optimal, but it works, we can always see if we can
  59         # do mmapping later on
  60         table = pq.read_table(self.path)
  61         self._load_table(table)
  62 
  63 vaex.file.other.dataset_type_map["arrow"] = DatasetArrow
  64 vaex.file.other.dataset_type_map["parquet"] = DatasetParquet
  65 
  66
  ```
  --- 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()
  
  ```
  
golden_diff: diff --git a/packages/vaex-arrow/vaex_arrow/dataset.py b/packages/vaex-arrow/vaex_arrow/dataset.py
  --- a/packages/vaex-arrow/vaex_arrow/dataset.py
  +++ b/packages/vaex-arrow/vaex_arrow/dataset.py
  @@ -28,14 +28,18 @@
           self._load_table(table)
       
       def _load_table(self, table):
  -        self._length_unfiltered =  self._length_original = table.num_rows
  -        self._index_end =  self._length_original = table.num_rows
  +        self._length_unfiltered = self._length_original = table.num_rows
  +        self._index_end = self._length_original = table.num_rows
           for col in table.columns:
               name = col.name
               # TODO: keep the arrow columns, and support and test chunks
               arrow_array = col.data.chunks[0]
  -            column = column_from_arrow_array(arrow_array)
  -
  +            if isinstance(arrow_array.type, pa.DictionaryType):
  +                column = column_from_arrow_array(arrow_array.indices)
  +                labels = column_from_arrow_array(arrow_array.dictionary).tolist()
  +                self._categories[name] = dict(labels=labels, N=len(labels))
  +            else:
  +                column = column_from_arrow_array(arrow_array)
               self.columns[name] = column
               self.column_names.append(name)
               self._save_assign_expression(name, vaex.expression.Expression(self, name))
  
verification_info: {"golden_diff": "diff --git a/packages/vaex-arrow/vaex_arrow/dataset.py b/packages/vaex-arrow/vaex_arrow/dataset.py\n--- a/packages/vaex-arrow/vaex_arrow/dataset.py\n+++ b/packages/vaex-arrow/vaex_arrow/dataset.py\n@@ -28,14 +28,18 @@\n         self._load_table(table)\n     \n     def _load_table(self, table):\n-        self._length_unfiltered =  self._length_original = table.num_rows\n-        self._index_end =  self._length_original = table.num_rows\n+        self._length_unfiltered = self._length_original = table.num_rows\n+        self._index_end = self._length_original = table.num_rows\n         for col in table.columns:\n             name = col.name\n             # TODO: keep the arrow columns, and support and test chunks\n             arrow_array = col.data.chunks[0]\n-            column = column_from_arrow_array(arrow_array)\n-\n+            if isinstance(arrow_array.type, pa.DictionaryType):\n+                column = column_from_arrow_array(arrow_array.indices)\n+                labels = column_from_arrow_array(arrow_array.dictionary).tolist()\n+                self._categories[name] = dict(labels=labels, N=len(labels))\n+            else:\n+                column = column_from_arrow_array(arrow_array)\n             self.columns[name] = column\n             self.column_names.append(name)\n             self._save_assign_expression(name, vaex.expression.Expression(self, name))\n", "issue": "Failing to open arrow stream with categoricals\nVaex fails to open arrow streams that contain categorical columns. It would be great if this was working as categoricals have given a great performance in many of my applications.\r\n\r\n```import pandas as pd\r\nimport pyarrow as pa\r\nimport numpy as np\r\nimport vaex\r\ndf = pd.DataFrame(\r\n    {\r\n        'col1': pd.Categorical.from_codes(np.full(1, 1), categories=['ABC', 'DEF'])\r\n    }\r\n)\r\ntable = pa.Table.from_pandas(df)\r\n\r\nwith pa.OSFile('test2.arrow', 'wb') as sink:\r\n    with pa.RecordBatchStreamWriter(sink, table.schema) as writer:\r\n        writer.write_table(table)\r\n\r\nwith pa.OSFile('test2.arrow', 'rb') as source:\r\n    df = pa.ipc.open_stream(source).read_pandas()\r\n\r\ndf = vaex.open('test2.arrow')\r\n```\r\n\r\nOutput:\r\n```\r\nERROR:MainThread:vaex:error opening 'test2.arrow'\r\n---------------------------------------------------------------------------\r\nNotImplementedError                       Traceback (most recent call last)\r\n in \r\n     17     df = pa.ipc.open_stream(source).read_pandas()\r\n     18 \r\n---> 19 df = vaex.open('test2.arrow')\r\n\r\nZ:\\Systemdateien\\Miniconda3\\envs\\finance\\lib\\site-packages\\vaex\\__init__.py in open(path, convert, shuffle, copy_index, *args, **kwargs)\r\n    189                         ds = from_csv(path, copy_index=copy_index, **kwargs)\r\n    190                     else:\r\n--> 191                         ds = vaex.file.open(path, *args, **kwargs)\r\n    192                     if convert and ds:\r\n    193                         ds.export_hdf5(filename_hdf5, shuffle=shuffle)\r\n\r\nZ:\\Systemdateien\\Miniconda3\\envs\\finance\\lib\\site-packages\\vaex\\file\\__init__.py in open(path, *args, **kwargs)\r\n     28     for opener in opener_classes:\r\n     29         if opener.can_open(path, *args, **kwargs):\r\n---> 30             return opener.open(path, *args, **kwargs)\r\n     31     if hdf5:\r\n     32         openers.extend(hdf5.dataset.dataset_type_map.items())\r\n\r\nZ:\\Systemdateien\\Miniconda3\\envs\\finance\\lib\\site-packages\\vaex_arrow\\opener.py in open(path, *args, **kwargs)\r\n      9     def open(path, *args, **kwargs):\r\n     10         from .dataset import DatasetArrow\r\n---> 11         return DatasetArrow(path, *args, **kwargs)\r\n     12 \r\n     13 class ParquetOpener:\r\n\r\nZ:\\Systemdateien\\Miniconda3\\envs\\finance\\lib\\site-packages\\vaex_arrow\\dataset.py in __init__(self, filename, table, write)\r\n     18         self._write = write\r\n     19         if table is None:\r\n---> 20             self._load()\r\n     21         else:\r\n     22             self._load_table(table)\r\n\r\nZ:\\Systemdateien\\Miniconda3\\envs\\finance\\lib\\site-packages\\vaex_arrow\\dataset.py in _load(self)\r\n     26         reader = pa.ipc.open_stream(source)\r\n     27         table = pa.Table.from_batches([b for b in reader])\r\n---> 28         self._load_table(table)\r\n     29 \r\n     30     def _load_table(self, table):\r\n\r\nZ:\\Systemdateien\\Miniconda3\\envs\\finance\\lib\\site-packages\\vaex_arrow\\dataset.py in _load_table(self, table)\r\n     35             # TODO: keep the arrow columns, and support and test chunks\r\n     36             arrow_array = col.data.chunks[0]\r\n---> 37             column = column_from_arrow_array(arrow_array)\r\n     38 \r\n     39             self.columns[name] = column\r\n\r\nZ:\\Systemdateien\\Miniconda3\\envs\\finance\\lib\\site-packages\\vaex_arrow\\convert.py in column_from_arrow_array(arrow_array)\r\n     28     buffers = arrow_array.buffers()\r\n     29     if len(buffers) == 2:\r\n---> 30         return numpy_array_from_arrow_array(arrow_array)\r\n     31     elif len(buffers) == 3 and  isinstance(arrow_array.type, type(pyarrow.string())):\r\n     32         bitmap_buffer, offsets, string_bytes = arrow_array.buffers()\r\n\r\nZ:\\Systemdateien\\Miniconda3\\envs\\finance\\lib\\site-packages\\vaex_arrow\\convert.py in numpy_array_from_arrow_array(arrow_array)\r\n     62         array = np.frombuffer(buffers[-1], dtype, len(arrow_array))# TODO: deal with offset ? [arrow_array.offset:arrow_array.offset + len(arrow_array)]\r\n     63     else:\r\n---> 64         dtype = arrow_array.type.to_pandas_dtype()\r\n     65     if np.bool_ == dtype:\r\n     66         # TODO: this will also be a copy, we probably want to support bitmasks as well\r\n\r\nZ:\\Systemdateien\\Miniconda3\\envs\\finance\\lib\\site-packages\\pyarrow\\types.pxi in pyarrow.lib.DataType.to_pandas_dtype()\r\n\r\nNotImplementedError: dictionary\r\n```\n", "before_files": [{"content": "__author__ = 'maartenbreddels'\nimport logging\n\nimport pyarrow as pa\nimport pyarrow.parquet as pq\n\nimport vaex.dataset\nimport vaex.file.other\nfrom .convert import column_from_arrow_array\nlogger = logging.getLogger(\"vaex_arrow\")\n\n\nclass DatasetArrow(vaex.dataset.DatasetLocal):\n    \"\"\"Implements storage using arrow\"\"\"\n\n    def __init__(self, filename=None, table=None, write=False):\n        super(DatasetArrow, self).__init__(name=filename, path=filename, column_names=[])\n        self._write = write\n        if table is None:\n            self._load()\n        else:\n            self._load_table(table)\n\n    def _load(self):\n        source = pa.memory_map(self.path)\n        reader = pa.ipc.open_stream(source)\n        table = pa.Table.from_batches([b for b in reader])\n        self._load_table(table)\n    \n    def _load_table(self, table):\n        self._length_unfiltered =  self._length_original = table.num_rows\n        self._index_end =  self._length_original = table.num_rows\n        for col in table.columns:\n            name = col.name\n            # TODO: keep the arrow columns, and support and test chunks\n            arrow_array = col.data.chunks[0]\n            column = column_from_arrow_array(arrow_array)\n\n            self.columns[name] = column\n            self.column_names.append(name)\n            self._save_assign_expression(name, vaex.expression.Expression(self, name))\n\n\n    @classmethod\n    def can_open(cls, path, *args, **kwargs):\n        return path.rpartition('.')[2] == 'arrow'\n\n    @classmethod\n    def get_options(cls, path):\n        return []\n\n    @classmethod\n    def option_to_args(cls, option):\n        return []\n\nclass DatasetParquet(DatasetArrow):\n    def _load(self):\n        # might not be optimal, but it works, we can always see if we can\n        # do mmapping later on\n        table = pq.read_table(self.path)\n        self._load_table(table)\n\nvaex.file.other.dataset_type_map[\"arrow\"] = DatasetArrow\nvaex.file.other.dataset_type_map[\"parquet\"] = DatasetParquet\n\n", "path": "packages/vaex-arrow/vaex_arrow/dataset.py"}], "after_files": [{"content": "__author__ = 'maartenbreddels'\nimport logging\n\nimport pyarrow as pa\nimport pyarrow.parquet as pq\n\nimport vaex.dataset\nimport vaex.file.other\nfrom .convert import column_from_arrow_array\nlogger = logging.getLogger(\"vaex_arrow\")\n\n\nclass DatasetArrow(vaex.dataset.DatasetLocal):\n    \"\"\"Implements storage using arrow\"\"\"\n\n    def __init__(self, filename=None, table=None, write=False):\n        super(DatasetArrow, self).__init__(name=filename, path=filename, column_names=[])\n        self._write = write\n        if table is None:\n            self._load()\n        else:\n            self._load_table(table)\n\n    def _load(self):\n        source = pa.memory_map(self.path)\n        reader = pa.ipc.open_stream(source)\n        table = pa.Table.from_batches([b for b in reader])\n        self._load_table(table)\n    \n    def _load_table(self, table):\n        self._length_unfiltered = self._length_original = table.num_rows\n        self._index_end = self._length_original = table.num_rows\n        for col in table.columns:\n            name = col.name\n            # TODO: keep the arrow columns, and support and test chunks\n            arrow_array = col.data.chunks[0]\n            if isinstance(arrow_array.type, pa.DictionaryType):\n                column = column_from_arrow_array(arrow_array.indices)\n                labels = column_from_arrow_array(arrow_array.dictionary).tolist()\n                self._categories[name] = dict(labels=labels, N=len(labels))\n            else:\n                column = column_from_arrow_array(arrow_array)\n            self.columns[name] = column\n            self.column_names.append(name)\n            self._save_assign_expression(name, vaex.expression.Expression(self, name))\n\n\n    @classmethod\n    def can_open(cls, path, *args, **kwargs):\n        return path.rpartition('.')[2] == 'arrow'\n\n    @classmethod\n    def get_options(cls, path):\n        return []\n\n    @classmethod\n    def option_to_args(cls, option):\n        return []\n\nclass DatasetParquet(DatasetArrow):\n    def _load(self):\n        # might not be optimal, but it works, we can always see if we can\n        # do mmapping later on\n        table = pq.read_table(self.path)\n        self._load_table(table)\n\nvaex.file.other.dataset_type_map[\"arrow\"] = DatasetArrow\nvaex.file.other.dataset_type_map[\"parquet\"] = DatasetParquet\n\n", "path": "packages/vaex-arrow/vaex_arrow/dataset.py"}]}