@@ -202,6 +202,141 @@ class RichRow:
202202 assert list (blosc2 .ctable_from_cframe (projected ).text [:]) == ["one" , None , "three" ]
203203
204204
205+ @pytest .fixture (params = ["taxi.parquet" , "taxi-data" ])
206+ def parquet_runtime (table_runtime , tmp_path , request ):
207+ import pyarrow as pa
208+ import pyarrow .parquet as pq
209+
210+ fs = fsspec .filesystem ("memory" )
211+ url = f"https://data.example/{ request .param }
212+ source = tmp_path / "taxi.parquet"
213+ pq .write_table (pa .table ({"fare" : [10 , 20 , 30 ], "cab" : ["a" , "b" , "c" ]}), source , row_group_size = 2 )
214+ fs .pipe_file (url , source .read_bytes ())
215+ path = table_runtime .with_name ("parquet-reference.b2z" )
216+ with blosc2 .RemoteCTable (
217+ url ,
218+ source_format = "parquet" ,
219+ _filesystem = fs ,
220+ cache_dir = tmp_path / "parquet-cache" ,
221+ columns = ["fare" , "cab" ],
222+ blosc2_batch_size = 2 ,
223+ ) as remote :
224+ assert list (remote .slice (0 , 2 ).fare [:]) == [10 , 20 ]
225+ remote .save (path )
226+ return path
227+
228+
229+ def test_remote_parquet_reference (parquet_runtime , tmp_path , monkeypatch ):
230+ import blosc2 .remote_parquet as parquet
231+ import pyarrow as pa
232+ import pyarrow .parquet as pq
233+
234+ from caterva2 .services import server
235+
236+ path = parquet_runtime
237+ manifest = remote_store .inspect (path )
238+ assert manifest ["parquet_caches" ]
239+ assert remote_store .root_kind (manifest ) == "ctable"
240+ q = server .quota_coordinator ()
241+ with monkeypatch .context () as patch :
242+ patch .setattr (
243+ parquet , "_open_source_handle" , lambda * args , ** kwargs : pytest .fail ("warm Parquet row fetched" )
244+ )
245+ assert srv_utils .read_metadata (path ).kind == "ctable"
246+ table = server .open_b2 (path , "@public/parquet-reference.b2z" )
247+ assert table .nrows == 3
248+ assert list (blosc2 .ctable_from_cframe (table .fetch (slice_ = slice (0 , 2 ))).fare [:]) == [10 , 20 ]
249+ assert q .usage ()["remote_cache_used" ] > 0
250+ assert remote_store .inspect (path )["parquet_caches" ] == []
251+ used = q .usage ()["used" ]
252+ with q .transaction () as db :
253+ db .execute ("UPDATE account SET cache_fill_suspended=1,quota=?" , (used ,))
254+ assert list (blosc2 .ctable_from_cframe (table .fetch (slice_ = slice (0 , 2 ))).fare [:]) == [10 , 20 ]
255+ with q .transaction () as db :
256+ db .execute ("UPDATE account SET cache_fill_suspended=0,quota=0" )
257+ for include_cache in (True , False ):
258+ export , _ , cleanup = q .remote .export_store (table .store , include_cache = include_cache )
259+ try :
260+ exported = remote_store .inspect (export )
261+ assert exported ["source" ] == manifest ["source" ]
262+ assert bool (exported ["parquet_caches" ]) == include_cache
263+ finally :
264+ cleanup ()
265+
266+ source = tmp_path / "taxi.parquet"
267+ pq .write_table (pa .table ({"fare" : [40 , 50 , 60 , 70 ], "cab" : ["d" , "e" , "f" , "g" ]}), source )
268+ fsspec .filesystem ("memory" ).pipe_file (manifest ["source" ]["urlpath" ], source .read_bytes ())
269+ assert list (blosc2 .ctable_from_cframe (table .fetch (slice_ = slice (0 , 2 ))).fare [:]) == [10 , 20 ]
270+ path .write_bytes (table .store .refreshed_bytes ())
271+ updated = server .open_b2 (path , "@public/parquet-reference.b2z" )
272+ assert list (blosc2 .ctable_from_cframe (updated .fetch (slice_ = slice (0 , 2 ))).fare [:]) == [40 , 50 ]
273+
274+
275+ @pytest .mark .parametrize ("policy" , ["none" , "memory" ])
276+ def test_remote_parquet_without_retention (parquet_runtime , policy ):
277+ from caterva2 .services import server
278+
279+ manifest = dict (
280+ remote_store .inspect (parquet_runtime ),
281+ cache_policy = policy ,
282+ max_cache_bytes = None if policy == "none" else 1 << 20 ,
283+ )
284+ path = parquet_runtime .with_name ("uncached.b2z" )
285+ remote_store .cold_export (manifest , path )
286+ table = server .open_b2 (path , "@public/uncached.b2z" )
287+ result = blosc2 .ctable_from_cframe (table .fetch (slice_ = slice (1 , 3 ), field = "cab" ))
288+ assert list (result .cab [:]) == ["b" , "c" ]
289+ assert server .quota_coordinator ().usage ()["remote_cache_used" ] == 0
290+ path .write_bytes (table .store .refreshed_bytes ())
291+ assert remote_store .inspect (path )["cache_policy" ] == policy
292+
293+
294+ @pytest .mark .asyncio
295+ async def test_remote_parquet_api (parquet_runtime , monkeypatch ):
296+ import httpx
297+
298+ from caterva2 .services import server
299+
300+ path = parquet_runtime
301+ monkeypatch .setattr (server .settings , "statedir" , path .parents [1 ])
302+ monkeypatch .setattr (server .settings , "public" , path .parent )
303+ monkeypatch .setitem (server .app .dependency_overrides , server .optional_user , lambda : None )
304+ async with httpx .AsyncClient (
305+ transport = httpx .ASGITransport (app = server .app ), base_url = "http://test"
306+ ) as client :
307+ response = await client .get ("/api/info/@public/parquet-reference.b2z" )
308+ assert response .status_code == 200 , response .text
309+ assert response .json ()["kind" ] == "ctable"
310+ response = await client .get (
311+ "/api/fetch/@public/parquet-reference.b2z" ,
312+ params = {"filter" : "fare >= 20" , "field" : "cab" , "slice_" : "0:1" },
313+ )
314+ assert response .status_code == 200 , response .text
315+ assert list (blosc2 .ctable_from_cframe (response .content ).cab [:]) == ["b" ]
316+ response = await client .post (
317+ "/htmx/path-view/@public/parquet-reference.b2z" , data = {"sortby" : "fare" }
318+ )
319+ assert response .status_code == 200 , response .text
320+ response = await client .get (
321+ "/api/download/@public/parquet-reference.b2z" , params = {"include_cache" : "false" }
322+ )
323+ assert response .status_code == 200 , response .text
324+ exported = path .with_name ("downloaded.b2z" )
325+ exported .write_bytes (response .content )
326+ assert remote_store .inspect (exported )["parquet_caches" ] == []
327+
328+
329+ def test_remote_parquet_resource_limit (parquet_runtime , monkeypatch ):
330+ import fastapi
331+
332+ from caterva2 .services import server
333+
334+ monkeypatch .setattr (remote_proxy , "policy" , dataclasses .replace (remote_proxy .policy , max_nbytes = 1 ))
335+ with pytest .raises (fastapi .HTTPException ) as exc :
336+ server .open_b2 (parquet_runtime , "@public/parquet-reference.b2z" )
337+ assert exc .value .status_code == 403
338+
339+
205340@pytest .mark .asyncio
206341async def test_standalone_remote_ctable_api (table_runtime , monkeypatch ):
207342 import httpx
0 commit comments