Feat/api parity

CHANGELOG.md

@@ -3,6 +3,33 @@
 ## Unreleased
 
 ### Breaking changes
+* `filter` takes a boolean expression (`Expr Bool -> DataFrame -> DataFrame`),
+  matching the lazy APIs. `filterWhere` is removed; use `filter`.
+* The old `filter :: Expr a -> (a -> Bool) -> …` is removed; use `filterBy`,
+  which takes the predicate first. The same applies to the typed API.
+* `DataFrame.Monad.filterWhereM` is renamed to `filterM`.
+* `DataFrame.Typed` mirrors `DataFrame`: expression combinators (`sum`,
+  `mean`, `count`, `lift`, `not`, casts, `asc`/`desc`, …) move to the new
+  `DataFrame.Typed.Functions`, imported like `DataFrame.Functions`.
+  `DataFrame.Typed` keeps `col`, `lit`, conditionals and operators, and now
+  exports the frame statistics (`mean`, `summarize`, `correlation`, …).
+* `DataFrame.Lazy` takes the eager shapes: `groupBy keys` then
+  `aggregate aggs`; `join jt [keys]` (multi-key) plus `innerJoin`,
+  `leftJoin`, `rightJoin`, `fullOuterJoin`; `sortBy` takes the eager
+  `SortOrder` (`Asc`/`Desc` expressions). `runDataFrame` is renamed `run`.
+  The lazy `SortOrder (Ascending | Descending)` type is removed.
+  `DataFrame.Lazy` re-exports the expression operators like `DataFrame`.
+* `DataFrame.Typed.Lazy` mirrors `DataFrame.Typed`: `sortBy` takes
+  `TSortOrder`, joins take a key list, `aggregate` takes an `as` builder, and
+  scans read their schema from the type (`scanCsv @Schema path`).
+  `fromTypedDataFrame` is removed (use `fromDataFrame`); expression
+  combinators come from `DataFrame.Typed.Functions`.
+* Typed eager `innerJoin` and `fullOuterJoin` and lazy INNER/FULL_OUTER joins
+  put left columns before right ones, matching eager joins and the typed
+  join schemas.
+* Typed `applyColumn` is renamed `apply`; typed `fromRecordsTyped` /
+  `toRecordsTyped` are exported from `DataFrame.Typed` as `fromRecords` /
+  `toRecords`.
 * Operators are identified by tag, not name. `UnaryOp`/`BinaryOp` expose
   `unaryId`/`binaryId :: OpRef …` (`Builtin` tag, `Named` text or
   `Anonymous` label); `unaryName`/`binaryName` are now plain functions.
@@ -23,12 +50,32 @@
   `DataFrame.Internal.Statistics`.
 
 ### Additions
-* `DataFrame.Typed` gains `cast`, `castWithDefault`, `castEither`,
+* `DataFrame.Lazy` and `DataFrame.Typed.Lazy` cover the eager API: row ops
+  (`filterBy`, `filterJust`, `drop`, `takeLast`, `range`, `topBy`,
+  `distinct`, `sample`, `shuffle`, splits and folds), column ops (`exclude`,
+  `rename`, `deriveMany`, `impute`, `apply*`, `insert*`, `cloneColumn`, …),
+  merges and set operations (`<>`/`append`, `|||`, `union`, `intersect`,
+  `difference`, `symmetricDifference`), summaries, and value-returning ops
+  (`mean`, `nRows`, `columnAs*`, `toCsv`, `writeCsv`, `writeParquet`,
+  `display`, …) that run the query and return `IO`. Row-local ops stream
+  batch by batch; ops that need the whole input collect it first.
+* `joinOn jt [(leftKey, rightKey)]` in all four APIs (typed:
+  `joinOn @'INNER @'[ '("id", "uid")]`); the result keeps the left names.
+* Typed parity with the untyped API: `topBy`, `deriveMany` (built with
+  `as`), `join @'INNER @'["key"]`, `applyInt`, `applyDouble`, `fold`,
+  `empty`, `null`, `(|>)`, display/rendering (`display`, `toCsv`,
+  `toMarkdown`, …), and IO re-exports from `DataFrame.Typed`.
+* Typed `writeParquet`, `writeParquetWithOptions`,
+  `readParquetFilesWithOpts`, `readSeparated`, `writeSeparated`, and the new
+  `DataFrame.Typed.IO.JSON` (`readJSON`, `readJSONEither`).
+* `DataFrame.Typed.Functions` gains `cast`, `castWithDefault`, `castEither`,
   `unsafeCast` (schema-checked column), `eq`, `lt`, `gt`, `leq`, `geq`,
   `and`, `or`, `countDistinct`, `topK`, `bottomK`, `isNull`, `isNotNull`,
   `contains` and `length`, completing parity with `DataFrame.Functions`.
 
 ### Fixes
+* Lazy CSV and Parquet scans return columns in file order, like eager
+  reads, instead of alphabetical order.
 * A user function named like a builtin (e.g. `"add"`) no longer runs the
   builtin's kernel instead of the user's function.
 * `eqExpr` no longer equates different `lift`/`lift2` functions, `fromMaybe`

README.md

@@ -235,7 +235,7 @@ $(D.declareColumnsFromCsvFile "./data/housing.csv")
 
 df <- D.readCsv "./data/housing.csv"
 df |> D.derive "rooms_per_household" (total_rooms / households)
-   |> D.filterWhere (median_income .>. 5)
+   |> D.filter (median_income .>. 5)
    |> D.groupBy ["ocean_proximity"]
    |> D.aggregate [F.mean median_house_value `as` "avg_value"]
    |> D.toMarkdown'
@@ -255,7 +255,7 @@ Compare this to the manual version which requires spelling out every column name
 -- Without TH — every column needs its name and type spelled out
 df |> D.derive "rooms_per_household"
         (F.col @Double "total_rooms" / F.col @Double "households")
-   |> D.filterWhere (F.col @Double "median_income" .>. F.lit 5)
+   |> D.filter (F.col @Double "median_income" .>. F.lit 5)
    |> D.take 5
    |> D.toMarkdown'
 ```
@@ -336,7 +336,7 @@ D.toRecords ordersDf :: Either Text [Order]
 And the typed bridge — `[Order]` to `TypedDataFrame OrderSchema` and back:
 
 ```haskell
-DT.thaw (DT.fromRecordsTyped xs :: DT.TypedDataFrame OrderSchema) |> D.toMarkdown'
+DT.thaw (DT.fromRecords xs :: DT.TypedDataFrame OrderSchema) |> D.toMarkdown'
 ```
 
 > <!-- scripths:mime text/plain -->
@@ -368,7 +368,7 @@ $(D.deriveSchemaValues ''Order)
 orders :: IO D.DataFrame
 orders = do
     raw <- D.readCsvWithSchema orderSchema "./data/orders.csv"
-    pure (D.filter orderAmount (> 100) raw)
+    pure (D.filterBy (> 100) orderAmount raw)
 ```
 
 > <!-- scripths:mime text/plain -->
@@ -417,7 +417,7 @@ case DT.freeze @EmployeeSchema employees of
     Nothing  -> "Schema mismatch!"
     Just tdf -> tdf
         |> DT.derive @"bonus" (DT.col @"salary" * DT.lit 0.1)
-        |> DT.filterWhere (DT.col @"salary" DT..>. DT.lit 50000)
+        |> DT.filter (DT.col @"salary" DT..>. DT.lit 50000)
         |> DT.select @'["name", "bonus"]
         |> DT.thaw
         |> D.toMarkdown'
@@ -435,11 +435,11 @@ case DT.freeze @EmployeeSchema employees of
 
 ```text
 -- Typo in column name -> compile error
-tdf |> DT.filterWhere (DT.col @"slary" DT..>. DT.lit 50000)
+tdf |> DT.filter (DT.col @"slary" DT..>. DT.lit 50000)
 -- error: Column 'slary' not found in schema
 
 -- Wrong type -> compile error
-tdf |> DT.filterWhere (DT.col @"name" DT..>. DT.lit 50000)
+tdf |> DT.filter (DT.col @"name" DT..>. DT.lit 50000)
 -- error: Couldn't match type 'Text' with 'Double'
 ```
 
@@ -468,7 +468,7 @@ DT.thaw (DT.filterAllJust stdf |> DT.derive @"scaled" (DT.col @"score" * DT.lit
 
 ## Lazy Queries
 
-For files too large to fit in memory, `DataFrame.Lazy` provides a streaming query engine. Declare a schema, build a query plan with the same familiar operations, and `runDataFrame` runs it through an optimizer before streaming results batch-by-batch:
+For files too large to fit in memory, `DataFrame.Lazy` provides a streaming query engine. Declare a schema, build a query plan with the same operations as the eager API, and `run` passes it through an optimizer before streaming results batch-by-batch:
 
 ```haskell
 import qualified DataFrame.Lazy as L
@@ -487,7 +487,7 @@ housingSchema = makeSchema
     , ("ocean_proximity",    schemaType @Text)
     ]
 
-lazyResult <- L.runDataFrame $
+lazyResult <- L.run $
     L.scanCsv housingSchema "./data/housing.csv"
     |> L.filter  (F.col @Double "median_income" .>. F.lit 5)
     |> L.derive  "value_per_income"
@@ -514,6 +514,23 @@ D.take 10 lazyResult |> D.toMarkdown'
 
 The optimizer pushes the filter into the scan, drops unreferenced columns before reading, and stops pulling batches once 1000 rows have been collected.
 
+### Switching between APIs
+
+The eager, typed, lazy and typed-lazy APIs share operation names and argument
+order, so moving a pipeline between them is mostly a change of imports:
+
+| | Frame ops (`D`) | Expressions (`F`) |
+|---|---|---|
+| Eager | `DataFrame` | `DataFrame.Functions` |
+| Typed | `DataFrame.Typed` | `DataFrame.Typed.Functions` |
+| Lazy | `DataFrame.Lazy` | `DataFrame.Functions` |
+| Typed lazy | `DataFrame.Typed.Lazy` | `DataFrame.Typed.Functions` |
+
+A lazy pipeline starts from `scanCsv`/`scanParquet`/`fromDataFrame` and ends with
+`run`; ops that return a value (`mean`, `nRows`, `writeCsv`, …) run the query
+and return `IO`. Typed APIs name columns at the type level (`col @"x"`,
+`select @'["x"]`) where untyped ones take text.
+
 ## Documentation
 
 * User guide: https://dataframe.readthedocs.io/en/latest/

app/Benchmark.hs

@@ -30,7 +30,7 @@ main = do
     let calculationTime = diffUTCTime endCalculation startCalculation
     putStrLn $ "Calculation Time: " ++ show calculationTime
     startFilter <- getCurrentTime
-    print $ D.filter (F.col @Double "0") (> 0.971) df D.|> D.take 10
+    print $ D.filterBy (> 0.971) (F.col @Double "0") df D.|> D.take 10
     endFilter <- getCurrentTime
     let filterTime = diffUTCTime endFilter startFilter
     putStrLn $ "Filter Time: " ++ show filterTime

app/LazyBenchmark.hs

@@ -186,26 +186,26 @@ main = do
                     ]
 
     runQuery "Q1 — preview first 20 rows (no filter)" $
-        L.runDataFrame $
+        L.run $
             L.take 20 $
                 L.scanCsv schema pathT
 
     runQuery "Q2 — filter (x > 0.999), limit 20" $
-        L.runDataFrame $
+        L.run $
             L.take 20 $
                 L.filter (col @Double "x" .> lit (0.999 :: Double)) $
                     L.scanCsv schema pathT
 
     runQuery "Q3 — filter (x > 0.999), derive z = x*y, select [id,z], limit 20" $
-        L.runDataFrame $
+        L.run $
             L.take 20 $
                 L.select ["id", "z"] $
                     L.derive "z" (col @Double "x" * col @Double "y") $
                         L.filter (col @Double "x" .> lit (0.999 :: Double)) $
                             L.scanCsv schema pathT
 
     runQuery "Q4 — filter fusion: (x > 0.5) . (y > 0.5), limit 20" $
-        L.runDataFrame $
+        L.run $
             L.take 20 $
                 L.filter (col @Double "y" .> lit (0.5 :: Double)) $
                     L.filter (col @Double "x" .> lit (0.5 :: Double)) $
@@ -216,7 +216,7 @@ main = do
             ++ approx (n `div` 1000)
             ++ " rows expected)"
         )
-        $ L.runDataFrame
+        $ L.run
         $ L.select ["id", "x"]
         $ L.filter (col @Double "x" .> lit (0.999 :: Double))
         $ L.scanCsv schema pathT

app/Synthesis.hs

@@ -13,6 +13,7 @@ import DataFrame.DecisionTree
 import DataFrame.Expression.Operators
 import qualified DataFrame.Functions as F
 import qualified DataFrame.Typed as DT
+import qualified DataFrame.Typed.Functions as DTF
 import System.Random
 
 $( DT.deriveSchemaFromCsvFileWith
@@ -106,9 +107,9 @@ clean ::
         )
 clean tdf =
     tdf
-        |> DT.replaceColumn @"Ticket" (DT.nullLift (T.filter isAlpha) (DT.col @"Ticket"))
-        |> DT.replaceColumn @"Name" (DT.nullLift extractTitle (DT.col @"Name"))
-        |> DT.replaceColumn @"Cabin" (DT.nullLift (T.take 1) (DT.col @"Cabin"))
+        |> DT.replaceColumn @"Ticket" (DTF.nullLift (T.filter isAlpha) (DT.col @"Ticket"))
+        |> DT.replaceColumn @"Name" (DTF.nullLift extractTitle (DT.col @"Name"))
+        |> DT.replaceColumn @"Cabin" (DTF.nullLift (T.take 1) (DT.col @"Cabin"))
         |> DT.renameMany
             @'[ '("Name", "title")
               , '("Cabin", "cabin_prefix")
@@ -133,7 +134,7 @@ computeAccuracy df =
                     df |> D.select ["Survived", "prediction"]
         survived = DT.col @"Survived"
         predCol = DT.col @"prediction"
-        count expr = fromIntegral (DT.nRows (DT.filterWhere expr tdf))
+        count expr = fromIntegral (DT.nRows (DT.filter expr tdf))
         tp = count ((survived DT..==. DT.lit 1) DT..&&. (predCol DT..==. DT.lit 1))
         tn = count ((survived DT..==. DT.lit 0) DT..&&. (predCol DT..==. DT.lit 0))
         fp = count ((survived DT..==. DT.lit 0) DT..&&. (predCol DT..==. DT.lit 1))

dataframe-arrow-bridge/src/DataFrame/IR.hs

@@ -58,7 +58,7 @@ import DataFrame.Operations.Core (insertVector, renameMany)
 import DataFrame.Operations.Join (JoinType (..), join)
 import DataFrame.Operations.Permutation (SortOrder (..), sortBy)
 import qualified DataFrame.Operations.Statistics as Stats
-import DataFrame.Operations.Subset (exclude, filterWhere, range, select)
+import DataFrame.Operations.Subset (exclude, range, select)
 import qualified DataFrame.Operations.Subset as Subset
 import DataFrame.Operations.Transformations (derive)
 import DataFrame.Schema (Schema, makeSchema, schemaType)
@@ -176,7 +176,7 @@ executePlan reader (Limit k node) =
 executePlan reader (Filter predJson node) = do
     df <- executePlan reader node
     case decodeExprAt @Bool predJson of
-        Right pred_ -> return $ filterWhere pred_ df
+        Right pred_ -> return $ Subset.filter pred_ df
         Left err -> ioError $ userError $ "DataFrame.IR.Filter: " <> err
 executePlan reader (Derive name exprJson node) = do
     df <- executePlan reader node
@@ -232,10 +232,10 @@ executePlan reader (WriteCsv path node) = do
     return df
 executePlan reader (ScanCsv path schemaPairs) = do
     schema <- buildSchema schemaPairs
-    Lazy.runDataFrame (Lazy.scanCsvWith reader schema (T.pack path))
+    Lazy.run (Lazy.scanCsvWith reader schema (T.pack path))
 executePlan _reader (ScanParquet path schemaPairs) = do
     schema <- buildSchema schemaPairs
-    Lazy.runDataFrame (Lazy.scanParquet schema (T.pack path))
+    Lazy.run (Lazy.scanParquet schema (T.pack path))
 
 -- | Build a SortOrder from a column's runtime type.
 mkSortOrder :: Bool -> T.Text -> Column -> SortOrder

dataframe-core/src-internal/DataFrame/Internal/Aggregation/Plan.hs

@@ -2,7 +2,6 @@
 {-# LANGUAGE ExplicitNamespaces #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
-{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 

dataframe-core/src-internal/DataFrame/Internal/Interpreter/Kernels.hs

@@ -1,6 +1,5 @@
 {-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE GADTs #-}
-{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}

dataframe-core/src/DataFrame/Typed/Schema.hs

@@ -34,6 +34,7 @@ module DataFrame.Typed.Schema (
     AssertAllPresent,
     AssertKeyTypesMatch,
     AssertDisjoint,
+    AssertAllAbsent,
     AssertAllColumnsHaveType,
     AssertRealColumn,
     AllColumnsReal,
@@ -201,6 +202,15 @@ type family ColumnNames (cols :: [(Symbol, Type)]) :: [Symbol] where
     ColumnNames '[] = '[]
     ColumnNames ('(n, _) ': rest) = n ': ColumnNames rest
 
+-- | Assert that none of the new columns already exists in the schema.
+type family
+    AssertAllAbsent (new :: [(Symbol, Type)]) (cols :: [(Symbol, Type)]) ::
+        Constraint
+    where
+    AssertAllAbsent '[] cols = ()
+    AssertAllAbsent ('(n, _) ': rest) cols =
+        (AssertAbsent n cols, AssertAllAbsent rest cols)
+
 -- | Assert that a column name is absent from the schema (for derive/insert).
 type family AssertAbsent (name :: Symbol) (cols :: [(Symbol, Type)]) :: Constraint where
     AssertAbsent name cols = AssertAbsentHelper name (HasName name cols) cols

dataframe-csv/src/DataFrame/Typed/IO/CSV.hs

@@ -8,8 +8,10 @@ module DataFrame.Typed.IO.CSV (
     readCsvWithError,
     readTsv,
     readCsvWithOpts,
+    readSeparated,
     writeCsv,
     writeTsv,
+    writeSeparated,
 ) where
 
 import Control.Applicative ((<|>))
@@ -76,3 +78,14 @@ writeCsv path = CSV.writeCsv path . thaw
 -- | Write a typed DataFrame to a tab-separated file.
 writeTsv :: FilePath -> TypedDataFrame cols -> IO ()
 writeTsv path = CSV.writeTsv path . thaw
+
+-- | Read a character-separated file with custom options; throws on mismatch.
+readSeparated ::
+    forall cols.
+    (KnownSchema cols, RuntimeSchema cols) =>
+    ReadOptions -> FilePath -> IO (TypedDataFrame cols)
+readSeparated = readCsvWithOpts @cols
+
+-- | Write a typed DataFrame with the given separator.
+writeSeparated :: Char -> FilePath -> TypedDataFrame cols -> IO ()
+writeSeparated sep path = CSV.writeSeparated sep path . thaw

dataframe-fusion/src/DataFrame/Fusion/Typed.hs

@@ -59,7 +59,6 @@ import Prelude hiding (filter, take)
 
 import qualified DataFrame.Internal.Column as IC
 import qualified DataFrame.Internal.Expression as IE
-import DataFrame.Lazy (SortOrder (..))
 import qualified DataFrame.Typed.Aggregate as AGG
 
 import qualified DataFrame.Fusion.FFI as F
@@ -218,6 +217,10 @@ joinWith how leftKey rightKey (DF leftPlan) (DF rightPlan) = do
                     ph <- runPlanOp (F.df_plan_join lp rp cHow cOn)
                     return (DF ph)
 
+-- | Sort direction for 'sortBy'.
+data SortOrder = Ascending | Descending
+    deriving (Show, Eq)
+
 -- | Sort the result by a list of (column, direction) pairs.
 sortBy ::
     [(T.Text, SortOrder)] ->

dataframe-json/dataframe-json.cabal

@@ -26,6 +26,7 @@ common warnings
 library
     import:             warnings
     exposed-modules:    DataFrame.IO.JSON
+                        DataFrame.Typed.IO.JSON
     build-depends:      base >= 4 && < 5,
                         aeson >= 0.11.0.0 && < 3,
                         bytestring >= 0.11 && < 0.14,