We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Allow lazy file opening for parquet formatted files
When I run `open` on an HDF5 file, the operation is done lazily – I believe that file handlers are opened but nothing is loaded into RAM. However running `open` on a parquet file results in eager evaluation, where data is immediately loaded into RAM. (Same goes for `open_many`, FWIW). This means that it's impossible to use vaex on large-than-RAM datasets if they're stored as parquet.

I've put a minimal working example below, although note it'll use like 3-6GB of RAM and dump like 5 GB to your HD:

```python
import numpy as np
import pandas as pd
import time
import vaex

# Creates a roughly 2.4 GB dataframe
pandas_df = pd.DataFrame({
    "numeric_1": np.random.rand(100000000),
    "numeric_2": np.random.rand(100000000),
    "numeric_3": np.random.rand(100000000)
})

# Saves to parquet and vaex-formatted hdf5
pandas_df.to_parquet("temp_df.parquet", index=False)
vaex_df = vaex.from_pandas(pandas_df)
vaex_df.export_hdf5("temp_df.hdf5")

del vaex_df
del pandas_df

start = time.time()
vaex_df_hdf5 = vaex.open("temp_df.hdf5")
print(f"Took {time.time() - start:.2f} seconds to open the hdf5 file")

start = time.time()
vaex_df_parquet = vaex.open("temp_df.parquet")
print(f"Took {time.time() - start:.2f} seconds to open the parquet file")
```

Output on my machine:
```
Took 0.01 seconds to open the hdf5 file
Took 7.89 seconds to open the parquet file
```

(You can also see that RAM usage is up, although printing that to screen from python is trickier 😄.)

I have no idea if it's possible, but it would be really awesome to get the same lazy opening for parquet files – A fair amount of the files I interact with are stored as parquet, and converting to HDF5 would be time consuming and annoying. 

--- 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-core/vaex/arrow/dataset.py`
Content:
```
1 __author__ = 'maartenbreddels'
2 import collections
3 import concurrent.futures
4 import logging
5 import multiprocessing
6 import os
7 
8 import pyarrow as pa
9 import pyarrow.dataset
10 
11 import vaex.dataset
12 import vaex.file.other
13 from ..itertools import buffer
14 
15 
16 logger = logging.getLogger("vaex.arrow.dataset")
17 
18 thread_count_default_io = os.environ.get('VAEX_NUM_THREADS_IO', multiprocessing.cpu_count() * 2 + 1)
19 thread_count_default_io = int(thread_count_default_io)
20 main_io_pool = None
21 
22 logger = logging.getLogger("vaex.multithreading")
23 
24 
25 def get_main_io_pool():
26     global main_io_pool
27     if main_io_pool is None:
28         main_io_pool = concurrent.futures.ThreadPoolExecutor(max_workers=thread_count_default_io)
29     return main_io_pool
30 
31 
32 class DatasetArrow(vaex.dataset.Dataset):
33     def __init__(self, ds):
34         super().__init__()
35         self._arrow_ds = ds
36         row_count = 0
37         for fragment in self._arrow_ds.get_fragments():
38             if hasattr(fragment, "ensure_complete_metadata"):
39                 fragment.ensure_complete_metadata()
40             for rg in fragment.row_groups:
41                 row_count += rg.num_rows
42         self._row_count = row_count
43         self._columns = {name: vaex.dataset.ColumnProxy(self, name, type) for name, type in
44                           zip(self._arrow_ds.schema.names, self._arrow_ds.schema.types)}
45         self._ids = {}
46 
47     def hashed(self):
48         raise NotImplementedError
49 
50     def slice(self, start, end):
51         # TODO: we can be smarter here, and trim off some fragments
52         if start == 0 and end == self.row_count:
53             return self
54         return vaex.dataset.DatasetSliced(self, start=start, end=end)
55 
56     def is_masked(self, column):
57         return False
58 
59     def shape(self, column):
60         return tuple()
61 
62     def __getitem__(self, item):
63         if isinstance(item, slice):
64             assert item.step in [1, None]
65             return vaex.dataset.DatasetSliced(self, item.start or 0, item.stop or self.row_count)
66         return self._columns[item]
67 
68     def close(self):
69         # no need to close it, it seem
70         pass
71 
72     def _chunk_producer(self, columns, chunk_size=None, reverse=False, start=0, end=None):
73         pool = get_main_io_pool()
74         offset = 0
75         for fragment_large in self._arrow_ds.get_fragments():
76             fragment_large_rows = sum([rg.num_rows for rg in fragment_large.row_groups])
77             # when do we want to split up? File size? max chunk size?
78             # if fragment_large_rows > chunk_size:
79             #     fragments = fragment_large.split_by_row_group()
80             # else:
81             #     # or not
82             #     fragments = [fragment_large]
83             import pyarrow.parquet
84             fragments = [fragment_large]
85             for fragment in fragments:
86                 rows = sum([rg.num_rows for rg in fragment.row_groups])
87                 chunk_start = offset
88                 chunk_end = offset + rows
89 
90                 length = chunk_end - chunk_start  # default length
91 
92                 if start >= chunk_end:  # we didn't find the beginning yet
93                     offset += length
94                     continue
95                 if end < chunk_start:  # we are past the end
96                     # assert False
97                     break
98                 def reader(fragment=fragment):
99                     table = fragment.to_table(columns=columns, use_threads=False)
100                     chunks = dict(zip(table.column_names, table.columns))
101                     return chunks
102 
103                 if start > chunk_start:
104                     # this means we have to cut off a piece of the beginning
105                     if end < chunk_end:
106                         # AND the end
107                         length = end - chunk_start  # without the start cut off
108                         length -= start - chunk_start  # correcting for the start cut off
109                         def slicer(chunk_start=chunk_start, reader=reader, length=length):
110                             chunks = reader()
111                             chunks = {name: ar.slice(start - chunk_start, length) for name, ar in chunks.items()}
112                             for name, ar in chunks.items():
113                                 assert len(ar) == length, f'Oops, array was expected to be of length {length} but was {len(ar)}'
114                             return chunks
115                         reader = slicer
116                     else:
117                         length -= start - chunk_start  # correcting for the start cut off
118                         def slicer(chunk_start=chunk_start, reader=reader, length=length):
119                             chunks = reader()
120                             chunks = {name: ar.slice(start - chunk_start) for name, ar in chunks.items()}
121                             for name, ar in chunks.items():
122                                 assert len(ar) == length, f'Oops, array was expected to be of length {length} but was {len(ar)}'
123                             return chunks
124                         reader = slicer
125                 else:
126                     if end < chunk_end:
127                         # we only need to cut off a piece of the end
128                         length = end - chunk_start
129                         def slicer(chunk_start=chunk_start, reader=reader, length=length):
130                             chunks = reader()
131                             chunks = {name: ar.slice(0, length) for name, ar in chunks.items()}
132                             for name, ar in chunks.items():
133                                 assert len(ar) == length, f'Oops, array was expected to be of length {length} but was {len(ar)}'
134                             return chunks
135                         reader = slicer
136                 offset += rows
137                 yield pool.submit(reader)
138 
139     def chunk_iterator(self, columns, chunk_size=None, reverse=False, start=0, end=None):
140         chunk_size = chunk_size or 1024*1024
141         i1 = 0
142         chunks_ready_list = []
143         i1 = i2 = 0
144 
145         for chunks_future in buffer(self._chunk_producer(columns, chunk_size, start=start, end=end or self._row_count), thread_count_default_io+3):
146             chunks = chunks_future.result()
147             chunks_ready_list.append(chunks)
148             total_row_count = sum([len(list(k.values())[0]) for k in chunks_ready_list])
149             if total_row_count > chunk_size:
150                 chunks_current_list, current_row_count = vaex.dataset._slice_of_chunks(chunks_ready_list, chunk_size)
151                 i2 += current_row_count
152                 yield i1, i2, vaex.dataset._concat_chunk_list(chunks_current_list)
153                 i1 = i2
154 
155         while chunks_ready_list:
156             chunks_current_list, current_row_count = vaex.dataset._slice_of_chunks(chunks_ready_list, chunk_size)
157             i2 += current_row_count
158             yield i1, i2, vaex.dataset._concat_chunk_list(chunks_current_list)
159             i1 = i2
160 
161 
162 
163 def from_table(table, as_numpy=False):
164     columns = dict(zip(table.schema.names, table.columns))
165     # TODO: this should be an DatasetArrow and/or DatasetParquet
166     dataset = vaex.dataset.DatasetArrays(columns)
167     df = vaex.dataframe.DataFrameLocal(dataset)
168     return df.as_numpy() if as_numpy else df
169 
170 
171 def open(filename, as_numpy=False):
172     source = pa.memory_map(filename)
173     try:
174         # first we try if it opens as stream
175         reader = pa.ipc.open_stream(source)
176     except pa.lib.ArrowInvalid:
177         # if not, we open as file
178         reader = pa.ipc.open_file(source)
179         # for some reason this reader is not iterable
180         batches = [reader.get_batch(i) for i in range(reader.num_record_batches)]
181     else:
182         # if a stream, we're good
183         batches = reader  # this reader is iterable
184     table = pa.Table.from_batches(batches)
185     return from_table(table, as_numpy=as_numpy)
186 
187 
188 def open_parquet(filename, as_numpy=False):
189     arrow_ds = pyarrow.dataset.dataset(filename)
190     ds = DatasetArrow(arrow_ds)
191     return vaex.from_dataset(ds)
192 
193 # vaex.file.other.dataset_type_map["arrow"] = DatasetArrow
194 # vaex.file.other.dataset_type_map["parquet"] = DatasetParquet
195 
196
```
--- 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/packages/vaex-core/vaex/arrow/dataset.py b/packages/vaex-core/vaex/arrow/dataset.py
--- a/packages/vaex-core/vaex/arrow/dataset.py
+++ b/packages/vaex-core/vaex/arrow/dataset.py
@@ -30,8 +30,9 @@
 
 
 class DatasetArrow(vaex.dataset.Dataset):
-    def __init__(self, ds):
+    def __init__(self, ds, max_rows_read=1024**2*10):
         super().__init__()
+        self.max_rows_read = max_rows_read
         self._arrow_ds = ds
         row_count = 0
         for fragment in self._arrow_ds.get_fragments():
@@ -70,18 +71,15 @@
         pass
 
     def _chunk_producer(self, columns, chunk_size=None, reverse=False, start=0, end=None):
+        import pyarrow.parquet
         pool = get_main_io_pool()
         offset = 0
         for fragment_large in self._arrow_ds.get_fragments():
             fragment_large_rows = sum([rg.num_rows for rg in fragment_large.row_groups])
-            # when do we want to split up? File size? max chunk size?
-            # if fragment_large_rows > chunk_size:
-            #     fragments = fragment_large.split_by_row_group()
-            # else:
-            #     # or not
-            #     fragments = [fragment_large]
-            import pyarrow.parquet
             fragments = [fragment_large]
+            # when do we want to split up? File size? max chunk size?
+            if fragment_large_rows > self.max_rows_read:
+                fragments = fragment_large.split_by_row_group()
             for fragment in fragments:
                 rows = sum([rg.num_rows for rg in fragment.row_groups])
                 chunk_start = offset

{"golden_diff": "diff --git a/packages/vaex-core/vaex/arrow/dataset.py b/packages/vaex-core/vaex/arrow/dataset.py\n--- a/packages/vaex-core/vaex/arrow/dataset.py\n+++ b/packages/vaex-core/vaex/arrow/dataset.py\n@@ -30,8 +30,9 @@\n \n \n class DatasetArrow(vaex.dataset.Dataset):\n-    def __init__(self, ds):\n+    def __init__(self, ds, max_rows_read=1024**2*10):\n         super().__init__()\n+        self.max_rows_read = max_rows_read\n         self._arrow_ds = ds\n         row_count = 0\n         for fragment in self._arrow_ds.get_fragments():\n@@ -70,18 +71,15 @@\n         pass\n \n     def _chunk_producer(self, columns, chunk_size=None, reverse=False, start=0, end=None):\n+        import pyarrow.parquet\n         pool = get_main_io_pool()\n         offset = 0\n         for fragment_large in self._arrow_ds.get_fragments():\n             fragment_large_rows = sum([rg.num_rows for rg in fragment_large.row_groups])\n-            # when do we want to split up? File size? max chunk size?\n-            # if fragment_large_rows > chunk_size:\n-            #     fragments = fragment_large.split_by_row_group()\n-            # else:\n-            #     # or not\n-            #     fragments = [fragment_large]\n-            import pyarrow.parquet\n             fragments = [fragment_large]\n+            # when do we want to split up? File size? max chunk size?\n+            if fragment_large_rows > self.max_rows_read:\n+                fragments = fragment_large.split_by_row_group()\n             for fragment in fragments:\n                 rows = sum([rg.num_rows for rg in fragment.row_groups])\n                 chunk_start = offset\n", "issue": "Allow lazy file opening for parquet formatted files\nWhen I run `open` on an HDF5 file, the operation is done lazily \u2013\u00a0I believe that file handlers are opened but nothing is loaded into RAM. However running `open` on a parquet file results in eager evaluation, where data is immediately loaded into RAM. (Same goes for `open_many`, FWIW). This means that it's impossible to use vaex on large-than-RAM datasets if they're stored as parquet.\r\n\r\nI've put a minimal working example below, although note it'll use like 3-6GB of RAM and dump like 5 GB to your HD:\r\n\r\n```python\r\nimport numpy as np\r\nimport pandas as pd\r\nimport time\r\nimport vaex\r\n\r\n# Creates a roughly 2.4 GB dataframe\r\npandas_df = pd.DataFrame({\r\n    \"numeric_1\": np.random.rand(100000000),\r\n    \"numeric_2\": np.random.rand(100000000),\r\n    \"numeric_3\": np.random.rand(100000000)\r\n})\r\n\r\n# Saves to parquet and vaex-formatted hdf5\r\npandas_df.to_parquet(\"temp_df.parquet\", index=False)\r\nvaex_df = vaex.from_pandas(pandas_df)\r\nvaex_df.export_hdf5(\"temp_df.hdf5\")\r\n\r\ndel vaex_df\r\ndel pandas_df\r\n\r\nstart = time.time()\r\nvaex_df_hdf5 = vaex.open(\"temp_df.hdf5\")\r\nprint(f\"Took {time.time() - start:.2f} seconds to open the hdf5 file\")\r\n\r\nstart = time.time()\r\nvaex_df_parquet = vaex.open(\"temp_df.parquet\")\r\nprint(f\"Took {time.time() - start:.2f} seconds to open the parquet file\")\r\n```\r\n\r\nOutput on my machine:\r\n```\r\nTook 0.01 seconds to open the hdf5 file\r\nTook 7.89 seconds to open the parquet file\r\n```\r\n\r\n(You can also see that RAM usage is up, although printing that to screen from python is trickier \ud83d\ude04.)\r\n\r\nI have no idea if it's possible, but it would be really awesome to get the same lazy opening for parquet files \u2013\u00a0A fair amount of the files I interact with are stored as parquet, and converting to HDF5 would be time consuming and annoying. \n", "before_files": [{"content": "__author__ = 'maartenbreddels'\nimport collections\nimport concurrent.futures\nimport logging\nimport multiprocessing\nimport os\n\nimport pyarrow as pa\nimport pyarrow.dataset\n\nimport vaex.dataset\nimport vaex.file.other\nfrom ..itertools import buffer\n\n\nlogger = logging.getLogger(\"vaex.arrow.dataset\")\n\nthread_count_default_io = os.environ.get('VAEX_NUM_THREADS_IO', multiprocessing.cpu_count() * 2 + 1)\nthread_count_default_io = int(thread_count_default_io)\nmain_io_pool = None\n\nlogger = logging.getLogger(\"vaex.multithreading\")\n\n\ndef get_main_io_pool():\n    global main_io_pool\n    if main_io_pool is None:\n        main_io_pool = concurrent.futures.ThreadPoolExecutor(max_workers=thread_count_default_io)\n    return main_io_pool\n\n\nclass DatasetArrow(vaex.dataset.Dataset):\n    def __init__(self, ds):\n        super().__init__()\n        self._arrow_ds = ds\n        row_count = 0\n        for fragment in self._arrow_ds.get_fragments():\n            if hasattr(fragment, \"ensure_complete_metadata\"):\n                fragment.ensure_complete_metadata()\n            for rg in fragment.row_groups:\n                row_count += rg.num_rows\n        self._row_count = row_count\n        self._columns = {name: vaex.dataset.ColumnProxy(self, name, type) for name, type in\n                          zip(self._arrow_ds.schema.names, self._arrow_ds.schema.types)}\n        self._ids = {}\n\n    def hashed(self):\n        raise NotImplementedError\n\n    def slice(self, start, end):\n        # TODO: we can be smarter here, and trim off some fragments\n        if start == 0 and end == self.row_count:\n            return self\n        return vaex.dataset.DatasetSliced(self, start=start, end=end)\n\n    def is_masked(self, column):\n        return False\n\n    def shape(self, column):\n        return tuple()\n\n    def __getitem__(self, item):\n        if isinstance(item, slice):\n            assert item.step in [1, None]\n            return vaex.dataset.DatasetSliced(self, item.start or 0, item.stop or self.row_count)\n        return self._columns[item]\n\n    def close(self):\n        # no need to close it, it seem\n        pass\n\n    def _chunk_producer(self, columns, chunk_size=None, reverse=False, start=0, end=None):\n        pool = get_main_io_pool()\n        offset = 0\n        for fragment_large in self._arrow_ds.get_fragments():\n            fragment_large_rows = sum([rg.num_rows for rg in fragment_large.row_groups])\n            # when do we want to split up? File size? max chunk size?\n            # if fragment_large_rows > chunk_size:\n            #     fragments = fragment_large.split_by_row_group()\n            # else:\n            #     # or not\n            #     fragments = [fragment_large]\n            import pyarrow.parquet\n            fragments = [fragment_large]\n            for fragment in fragments:\n                rows = sum([rg.num_rows for rg in fragment.row_groups])\n                chunk_start = offset\n                chunk_end = offset + rows\n\n                length = chunk_end - chunk_start  # default length\n\n                if start >= chunk_end:  # we didn't find the beginning yet\n                    offset += length\n                    continue\n                if end < chunk_start:  # we are past the end\n                    # assert False\n                    break\n                def reader(fragment=fragment):\n                    table = fragment.to_table(columns=columns, use_threads=False)\n                    chunks = dict(zip(table.column_names, table.columns))\n                    return chunks\n\n                if start > chunk_start:\n                    # this means we have to cut off a piece of the beginning\n                    if end < chunk_end:\n                        # AND the end\n                        length = end - chunk_start  # without the start cut off\n                        length -= start - chunk_start  # correcting for the start cut off\n                        def slicer(chunk_start=chunk_start, reader=reader, length=length):\n                            chunks = reader()\n                            chunks = {name: ar.slice(start - chunk_start, length) for name, ar in chunks.items()}\n                            for name, ar in chunks.items():\n                                assert len(ar) == length, f'Oops, array was expected to be of length {length} but was {len(ar)}'\n                            return chunks\n                        reader = slicer\n                    else:\n                        length -= start - chunk_start  # correcting for the start cut off\n                        def slicer(chunk_start=chunk_start, reader=reader, length=length):\n                            chunks = reader()\n                            chunks = {name: ar.slice(start - chunk_start) for name, ar in chunks.items()}\n                            for name, ar in chunks.items():\n                                assert len(ar) == length, f'Oops, array was expected to be of length {length} but was {len(ar)}'\n                            return chunks\n                        reader = slicer\n                else:\n                    if end < chunk_end:\n                        # we only need to cut off a piece of the end\n                        length = end - chunk_start\n                        def slicer(chunk_start=chunk_start, reader=reader, length=length):\n                            chunks = reader()\n                            chunks = {name: ar.slice(0, length) for name, ar in chunks.items()}\n                            for name, ar in chunks.items():\n                                assert len(ar) == length, f'Oops, array was expected to be of length {length} but was {len(ar)}'\n                            return chunks\n                        reader = slicer\n                offset += rows\n                yield pool.submit(reader)\n\n    def chunk_iterator(self, columns, chunk_size=None, reverse=False, start=0, end=None):\n        chunk_size = chunk_size or 1024*1024\n        i1 = 0\n        chunks_ready_list = []\n        i1 = i2 = 0\n\n        for chunks_future in buffer(self._chunk_producer(columns, chunk_size, start=start, end=end or self._row_count), thread_count_default_io+3):\n            chunks = chunks_future.result()\n            chunks_ready_list.append(chunks)\n            total_row_count = sum([len(list(k.values())[0]) for k in chunks_ready_list])\n            if total_row_count > chunk_size:\n                chunks_current_list, current_row_count = vaex.dataset._slice_of_chunks(chunks_ready_list, chunk_size)\n                i2 += current_row_count\n                yield i1, i2, vaex.dataset._concat_chunk_list(chunks_current_list)\n                i1 = i2\n\n        while chunks_ready_list:\n            chunks_current_list, current_row_count = vaex.dataset._slice_of_chunks(chunks_ready_list, chunk_size)\n            i2 += current_row_count\n            yield i1, i2, vaex.dataset._concat_chunk_list(chunks_current_list)\n            i1 = i2\n\n\n\ndef from_table(table, as_numpy=False):\n    columns = dict(zip(table.schema.names, table.columns))\n    # TODO: this should be an DatasetArrow and/or DatasetParquet\n    dataset = vaex.dataset.DatasetArrays(columns)\n    df = vaex.dataframe.DataFrameLocal(dataset)\n    return df.as_numpy() if as_numpy else df\n\n\ndef open(filename, as_numpy=False):\n    source = pa.memory_map(filename)\n    try:\n        # first we try if it opens as stream\n        reader = pa.ipc.open_stream(source)\n    except pa.lib.ArrowInvalid:\n        # if not, we open as file\n        reader = pa.ipc.open_file(source)\n        # for some reason this reader is not iterable\n        batches = [reader.get_batch(i) for i in range(reader.num_record_batches)]\n    else:\n        # if a stream, we're good\n        batches = reader  # this reader is iterable\n    table = pa.Table.from_batches(batches)\n    return from_table(table, as_numpy=as_numpy)\n\n\ndef open_parquet(filename, as_numpy=False):\n    arrow_ds = pyarrow.dataset.dataset(filename)\n    ds = DatasetArrow(arrow_ds)\n    return vaex.from_dataset(ds)\n\n# vaex.file.other.dataset_type_map[\"arrow\"] = DatasetArrow\n# vaex.file.other.dataset_type_map[\"parquet\"] = DatasetParquet\n\n", "path": "packages/vaex-core/vaex/arrow/dataset.py"}], "after_files": [{"content": "__author__ = 'maartenbreddels'\nimport collections\nimport concurrent.futures\nimport logging\nimport multiprocessing\nimport os\n\nimport pyarrow as pa\nimport pyarrow.dataset\n\nimport vaex.dataset\nimport vaex.file.other\nfrom ..itertools import buffer\n\n\nlogger = logging.getLogger(\"vaex.arrow.dataset\")\n\nthread_count_default_io = os.environ.get('VAEX_NUM_THREADS_IO', multiprocessing.cpu_count() * 2 + 1)\nthread_count_default_io = int(thread_count_default_io)\nmain_io_pool = None\n\nlogger = logging.getLogger(\"vaex.multithreading\")\n\n\ndef get_main_io_pool():\n    global main_io_pool\n    if main_io_pool is None:\n        main_io_pool = concurrent.futures.ThreadPoolExecutor(max_workers=thread_count_default_io)\n    return main_io_pool\n\n\nclass DatasetArrow(vaex.dataset.Dataset):\n    def __init__(self, ds, max_rows_read=1024**2*10):\n        super().__init__()\n        self.max_rows_read = max_rows_read\n        self._arrow_ds = ds\n        row_count = 0\n        for fragment in self._arrow_ds.get_fragments():\n            if hasattr(fragment, \"ensure_complete_metadata\"):\n                fragment.ensure_complete_metadata()\n            for rg in fragment.row_groups:\n                row_count += rg.num_rows\n        self._row_count = row_count\n        self._columns = {name: vaex.dataset.ColumnProxy(self, name, type) for name, type in\n                          zip(self._arrow_ds.schema.names, self._arrow_ds.schema.types)}\n        self._ids = {}\n\n    def hashed(self):\n        raise NotImplementedError\n\n    def slice(self, start, end):\n        # TODO: we can be smarter here, and trim off some fragments\n        if start == 0 and end == self.row_count:\n            return self\n        return vaex.dataset.DatasetSliced(self, start=start, end=end)\n\n    def is_masked(self, column):\n        return False\n\n    def shape(self, column):\n        return tuple()\n\n    def __getitem__(self, item):\n        if isinstance(item, slice):\n            assert item.step in [1, None]\n            return vaex.dataset.DatasetSliced(self, item.start or 0, item.stop or self.row_count)\n        return self._columns[item]\n\n    def close(self):\n        # no need to close it, it seem\n        pass\n\n    def _chunk_producer(self, columns, chunk_size=None, reverse=False, start=0, end=None):\n        import pyarrow.parquet\n        pool = get_main_io_pool()\n        offset = 0\n        for fragment_large in self._arrow_ds.get_fragments():\n            fragment_large_rows = sum([rg.num_rows for rg in fragment_large.row_groups])\n            fragments = [fragment_large]\n            # when do we want to split up? File size? max chunk size?\n            if fragment_large_rows > self.max_rows_read:\n                fragments = fragment_large.split_by_row_group()\n            for fragment in fragments:\n                rows = sum([rg.num_rows for rg in fragment.row_groups])\n                chunk_start = offset\n                chunk_end = offset + rows\n\n                length = chunk_end - chunk_start  # default length\n\n                if start >= chunk_end:  # we didn't find the beginning yet\n                    offset += length\n                    continue\n                if end < chunk_start:  # we are past the end\n                    # assert False\n                    break\n                def reader(fragment=fragment):\n                    table = fragment.to_table(columns=columns, use_threads=False)\n                    chunks = dict(zip(table.column_names, table.columns))\n                    return chunks\n\n                if start > chunk_start:\n                    # this means we have to cut off a piece of the beginning\n                    if end < chunk_end:\n                        # AND the end\n                        length = end - chunk_start  # without the start cut off\n                        length -= start - chunk_start  # correcting for the start cut off\n                        def slicer(chunk_start=chunk_start, reader=reader, length=length):\n                            chunks = reader()\n                            chunks = {name: ar.slice(start - chunk_start, length) for name, ar in chunks.items()}\n                            for name, ar in chunks.items():\n                                assert len(ar) == length, f'Oops, array was expected to be of length {length} but was {len(ar)}'\n                            return chunks\n                        reader = slicer\n                    else:\n                        length -= start - chunk_start  # correcting for the start cut off\n                        def slicer(chunk_start=chunk_start, reader=reader, length=length):\n                            chunks = reader()\n                            chunks = {name: ar.slice(start - chunk_start) for name, ar in chunks.items()}\n                            for name, ar in chunks.items():\n                                assert len(ar) == length, f'Oops, array was expected to be of length {length} but was {len(ar)}'\n                            return chunks\n                        reader = slicer\n                else:\n                    if end < chunk_end:\n                        # we only need to cut off a piece of the end\n                        length = end - chunk_start\n                        def slicer(chunk_start=chunk_start, reader=reader, length=length):\n                            chunks = reader()\n                            chunks = {name: ar.slice(0, length) for name, ar in chunks.items()}\n                            for name, ar in chunks.items():\n                                assert len(ar) == length, f'Oops, array was expected to be of length {length} but was {len(ar)}'\n                            return chunks\n                        reader = slicer\n                offset += rows\n                yield pool.submit(reader)\n\n    def chunk_iterator(self, columns, chunk_size=None, reverse=False, start=0, end=None):\n        chunk_size = chunk_size or 1024*1024\n        i1 = 0\n        chunks_ready_list = []\n        i1 = i2 = 0\n\n        for chunks_future in buffer(self._chunk_producer(columns, chunk_size, start=start, end=end or self._row_count), thread_count_default_io+3):\n            chunks = chunks_future.result()\n            chunks_ready_list.append(chunks)\n            total_row_count = sum([len(list(k.values())[0]) for k in chunks_ready_list])\n            if total_row_count > chunk_size:\n                chunks_current_list, current_row_count = vaex.dataset._slice_of_chunks(chunks_ready_list, chunk_size)\n                i2 += current_row_count\n                yield i1, i2, vaex.dataset._concat_chunk_list(chunks_current_list)\n                i1 = i2\n\n        while chunks_ready_list:\n            chunks_current_list, current_row_count = vaex.dataset._slice_of_chunks(chunks_ready_list, chunk_size)\n            i2 += current_row_count\n            yield i1, i2, vaex.dataset._concat_chunk_list(chunks_current_list)\n            i1 = i2\n\n\n\ndef from_table(table, as_numpy=False):\n    columns = dict(zip(table.schema.names, table.columns))\n    # TODO: this should be an DatasetArrow and/or DatasetParquet\n    dataset = vaex.dataset.DatasetArrays(columns)\n    df = vaex.dataframe.DataFrameLocal(dataset)\n    return df.as_numpy() if as_numpy else df\n\n\ndef open(filename, as_numpy=False):\n    source = pa.memory_map(filename)\n    try:\n        # first we try if it opens as stream\n        reader = pa.ipc.open_stream(source)\n    except pa.lib.ArrowInvalid:\n        # if not, we open as file\n        reader = pa.ipc.open_file(source)\n        # for some reason this reader is not iterable\n        batches = [reader.get_batch(i) for i in range(reader.num_record_batches)]\n    else:\n        # if a stream, we're good\n        batches = reader  # this reader is iterable\n    table = pa.Table.from_batches(batches)\n    return from_table(table, as_numpy=as_numpy)\n\n\ndef open_parquet(filename, as_numpy=False):\n    arrow_ds = pyarrow.dataset.dataset(filename)\n    ds = DatasetArrow(arrow_ds)\n    return vaex.from_dataset(ds)\n\n# vaex.file.other.dataset_type_map[\"arrow\"] = DatasetArrow\n# vaex.file.other.dataset_type_map[\"parquet\"] = DatasetParquet\n\n", "path": "packages/vaex-core/vaex/arrow/dataset.py"}]}