dataframe-lazy 1.0.2.0 → 1.1.0.0
raw patch · 2 files changed
+22/−59 lines, 2 filesdep ~dataframe-parquetPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: dataframe-parquet
API changes (from Hackage documentation)
Files
dataframe-lazy.cabal view
@@ -1,6 +1,6 @@ cabal-version: 2.4 name: dataframe-lazy-version: 1.0.2.0+version: 1.1.0.0 synopsis: Lazy query engine for the dataframe ecosystem. description: The lazy/streaming query engine: relational-algebra plans, optimizer,@@ -45,14 +45,14 @@ dataframe-core ^>= 1.1, dataframe-csv ^>= 1.0.2, dataframe-operations ^>= 1.1.1,- dataframe-parquet ^>= 1.0,+ dataframe-parquet ^>= 1.1, dataframe-parsing ^>= 1.0.2, directory >= 1.3.0.0 && < 2, filepath >= 1.4 && < 2, Glob >= 0.10 && < 1, stm >= 2.5 && < 3, temporary >= 1.3 && < 2,- text >= 2.0 && < 3,+ text >= 2.1 && < 3, vector ^>= 0.13 hs-source-dirs: src default-language: Haskell2010
src/DataFrame/Lazy/Internal/Executor.hs view
@@ -26,10 +26,11 @@ grouping). This inherits the eager Rounds 5-10 gains and produces results byte-identical to the eager path. -When a source is UNBOUNDED (HuggingFace / online Parquet), the per-batch+When a source is UNBOUNDED (an online/streaming source), the per-batch streaming partial-aggregation / streaming-probe paths are preserved so the query runs in constant memory. Boundedness is read off the physical plan-leaves by 'isBounded'.+leaves by 'isBounded'. All current scan leaves are local files, so the+streaming paths are a latent capability with no in-tree producer. -} module DataFrame.Lazy.Internal.Executor ( CsvReader,@@ -133,13 +134,14 @@ (local CSV or local Parquet files): the whole result can be materialised, so blocking/bounded operators route through the whole-frame fast eager ops. -A plan is UNBOUNDED when any leaf is an online/streaming source (a HuggingFace-Parquet URI), in which case the streaming partial-aggregation / streaming-probe-paths are kept so the query runs in constant memory.+A plan is UNBOUNDED when any leaf is an online/streaming source, in which case+the streaming partial-aggregation / streaming-probe paths are kept so the query+runs in constant memory. No in-tree scan produces an unbounded source today,+so this always holds; the routing is retained for future streaming sources. -} isBounded :: PhysicalPlan -> Bool isBounded (PhysicalScan (CsvSource{}) _) = True-isBounded (PhysicalScan (ParquetSource path) _) = not (Parquet.isHFUri path)+isBounded (PhysicalScan (ParquetSource _) _) = True isBounded (PhysicalProject _ c) = isBounded c isBounded (PhysicalFilter _ c) = isBounded c isBounded (PhysicalDerive _ _ c) = isBounded c@@ -539,65 +541,26 @@ via 'ParquetReadOptions'. -} executeParquetScan :: FilePath -> ScanConfig -> IO Stream-executeParquetScan path cfg- | Parquet.isHFUri path = executeHFParquetScan path cfg- | otherwise = do- isDir <- doesDirectoryExist path- let pat = if isDir then path </> "*" else path- matches <- glob pat- files <- filterM (fmap not . doesDirectoryExist) matches- when (null files) $- error ("executeParquetScan: no parquet files found for " ++ path)- let opts =- Parquet.defaultParquetReadOptions- { Parquet.selectedColumns = Just (M.keys (elements (scanSchema cfg)))- , Parquet.predicate = scanPushdownPredicate cfg- }- ref <- newIORef files- return . Stream $ do- fs <- readIORef ref- case fs of- [] -> return Nothing- (f : rest) -> do- writeIORef ref rest- Just <$> Parquet.readParquetWithOpts opts f--{- | HuggingFace Parquet scan. Files are resolved once (API call or direct URL)-then downloaded one at a time as the stream is pulled — so only one file's worth-of data is in memory at a time, regardless of dataset size.--}---- TODO: mchavinda - this should be a more general online file scanner.-executeHFParquetScan :: FilePath -> ScanConfig -> IO Stream-executeHFParquetScan path cfg = do- ref <- case Parquet.parseHFUri path of- Left err -> error err- Right r -> pure r- mToken <- Parquet.getHFToken- hfFiles <-- if Parquet.hasGlob (Parquet.hfGlob ref)- then Parquet.resolveHFUrls mToken ref- else do- let url = Parquet.directHFUrl ref- filename = last $ T.splitOn "/" (Parquet.hfGlob ref)- pure [Parquet.HFParquetFile url "" "" filename]- when (null hfFiles) $- error ("executeParquetScan: no HF parquet files found for " ++ path)+executeParquetScan path cfg = do+ isDir <- doesDirectoryExist path+ let pat = if isDir then path </> "*" else path+ matches <- glob pat+ files <- filterM (fmap not . doesDirectoryExist) matches+ when (null files) $+ error ("executeParquetScan: no parquet files found for " ++ path) let opts = Parquet.defaultParquetReadOptions { Parquet.selectedColumns = Just (M.keys (elements (scanSchema cfg))) , Parquet.predicate = scanPushdownPredicate cfg }- filesRef <- newIORef hfFiles+ ref <- newIORef files return . Stream $ do- fs <- readIORef filesRef+ fs <- readIORef ref case fs of [] -> return Nothing (f : rest) -> do- writeIORef filesRef rest- -- Download a single file, read it, then return the batch.- [localPath] <- Parquet.downloadHFFiles mToken [f]- Just <$> Parquet.readParquetWithOpts opts localPath+ writeIORef ref rest+ Just <$> Parquet.readParquetWithOpts opts f -- --------------------------------------------------------------------------- -- CSV scan implementation