packages feed

krapsh (empty) → 0.1.6.0

raw patch · 64 files changed

+6972/−0 lines, 64 filesdep +QuickCheckdep +SHAdep +aesonsetup-changed

Dependencies added: QuickCheck, SHA, aeson, aeson-pretty, base, base16-bytestring, binary, bytestring, containers, cryptohash-sha256, deepseq, exceptions, formatting, hashable, hspec, krapsh, lens, monad-logger, mtl, random, raw-strings-qq, scientific, text, text-format, transformers, unordered-containers, vector, wreq

Files

+ LICENSE view
@@ -0,0 +1,201 @@+                                 Apache License+                           Version 2.0, January 2004+                        http://www.apache.org/licenses/++   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION++   1. Definitions.++      "License" shall mean the terms and conditions for use, reproduction,+      and distribution as defined by Sections 1 through 9 of this document.++      "Licensor" shall mean the copyright owner or entity authorized by+      the copyright owner that is granting the License.++      "Legal Entity" shall mean the union of the acting entity and all+      other entities that control, are controlled by, or are under common+      control with that entity. For the purposes of this definition,+      "control" means (i) the power, direct or indirect, to cause the+      direction or management of such entity, whether by contract or+      otherwise, or (ii) ownership of fifty percent (50%) or more of the+      outstanding shares, or (iii) beneficial ownership of such entity.++      "You" (or "Your") shall mean an individual or Legal Entity+      exercising permissions granted by this License.++      "Source" form shall mean the preferred form for making modifications,+      including but not limited to software source code, documentation+      source, and configuration files.++      "Object" form shall mean any form resulting from mechanical+      transformation or translation of a Source form, including but+      not limited to compiled object code, generated documentation,+      and conversions to other media types.++      "Work" shall mean the work of authorship, whether in Source or+      Object form, made available under the License, as indicated by a+      copyright notice that is included in or attached to the work+      (an example is provided in the Appendix below).++      "Derivative Works" shall mean any work, whether in Source or Object+      form, that is based on (or derived from) the Work and for which the+      editorial revisions, annotations, elaborations, or other modifications+      represent, as a whole, an original work of authorship. For the purposes+      of this License, Derivative Works shall not include works that remain+      separable from, or merely link (or bind by name) to the interfaces of,+      the Work and Derivative Works thereof.++      "Contribution" shall mean any work of authorship, including+      the original version of the Work and any modifications or additions+      to that Work or Derivative Works thereof, that is intentionally+      submitted to Licensor for inclusion in the Work by the copyright owner+      or by an individual or Legal Entity authorized to submit on behalf of+      the copyright owner. For the purposes of this definition, "submitted"+      means any form of electronic, verbal, or written communication sent+      to the Licensor or its representatives, including but not limited to+      communication on electronic mailing lists, source code control systems,+      and issue tracking systems that are managed by, or on behalf of, the+      Licensor for the purpose of discussing and improving the Work, but+      excluding communication that is conspicuously marked or otherwise+      designated in writing by the copyright owner as "Not a Contribution."++      "Contributor" shall mean Licensor and any individual or Legal Entity+      on behalf of whom a Contribution has been received by Licensor and+      subsequently incorporated within the Work.++   2. Grant of Copyright License. Subject to the terms and conditions of+      this License, each Contributor hereby grants to You a perpetual,+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable+      copyright license to reproduce, prepare Derivative Works of,+      publicly display, publicly perform, sublicense, and distribute the+      Work and such Derivative Works in Source or Object form.++   3. Grant of Patent License. Subject to the terms and conditions of+      this License, each Contributor hereby grants to You a perpetual,+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable+      (except as stated in this section) patent license to make, have made,+      use, offer to sell, sell, import, and otherwise transfer the Work,+      where such license applies only to those patent claims licensable+      by such Contributor that are necessarily infringed by their+      Contribution(s) alone or by combination of their Contribution(s)+      with the Work to which such Contribution(s) was submitted. If You+      institute patent litigation against any entity (including a+      cross-claim or counterclaim in a lawsuit) alleging that the Work+      or a Contribution incorporated within the Work constitutes direct+      or contributory patent infringement, then any patent licenses+      granted to You under this License for that Work shall terminate+      as of the date such litigation is filed.++   4. Redistribution. You may reproduce and distribute copies of the+      Work or Derivative Works thereof in any medium, with or without+      modifications, and in Source or Object form, provided that You+      meet the following conditions:++      (a) You must give any other recipients of the Work or+          Derivative Works a copy of this License; and++      (b) You must cause any modified files to carry prominent notices+          stating that You changed the files; and++      (c) You must retain, in the Source form of any Derivative Works+          that You distribute, all copyright, patent, trademark, and+          attribution notices from the Source form of the Work,+          excluding those notices that do not pertain to any part of+          the Derivative Works; and++      (d) If the Work includes a "NOTICE" text file as part of its+          distribution, then any Derivative Works that You distribute must+          include a readable copy of the attribution notices contained+          within such NOTICE file, excluding those notices that do not+          pertain to any part of the Derivative Works, in at least one+          of the following places: within a NOTICE text file distributed+          as part of the Derivative Works; within the Source form or+          documentation, if provided along with the Derivative Works; or,+          within a display generated by the Derivative Works, if and+          wherever such third-party notices normally appear. The contents+          of the NOTICE file are for informational purposes only and+          do not modify the License. You may add Your own attribution+          notices within Derivative Works that You distribute, alongside+          or as an addendum to the NOTICE text from the Work, provided+          that such additional attribution notices cannot be construed+          as modifying the License.++      You may add Your own copyright statement to Your modifications and+      may provide additional or different license terms and conditions+      for use, reproduction, or distribution of Your modifications, or+      for any such Derivative Works as a whole, provided Your use,+      reproduction, and distribution of the Work otherwise complies with+      the conditions stated in this License.++   5. Submission of Contributions. Unless You explicitly state otherwise,+      any Contribution intentionally submitted for inclusion in the Work+      by You to the Licensor shall be under the terms and conditions of+      this License, without any additional terms or conditions.+      Notwithstanding the above, nothing herein shall supersede or modify+      the terms of any separate license agreement you may have executed+      with Licensor regarding such Contributions.++   6. Trademarks. This License does not grant permission to use the trade+      names, trademarks, service marks, or product names of the Licensor,+      except as required for reasonable and customary use in describing the+      origin of the Work and reproducing the content of the NOTICE file.++   7. Disclaimer of Warranty. Unless required by applicable law or+      agreed to in writing, Licensor provides the Work (and each+      Contributor provides its Contributions) on an "AS IS" BASIS,+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or+      implied, including, without limitation, any warranties or conditions+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A+      PARTICULAR PURPOSE. You are solely responsible for determining the+      appropriateness of using or redistributing the Work and assume any+      risks associated with Your exercise of permissions under this License.++   8. Limitation of Liability. In no event and under no legal theory,+      whether in tort (including negligence), contract, or otherwise,+      unless required by applicable law (such as deliberate and grossly+      negligent acts) or agreed to in writing, shall any Contributor be+      liable to You for damages, including any direct, indirect, special,+      incidental, or consequential damages of any character arising as a+      result of this License or out of the use or inability to use the+      Work (including but not limited to damages for loss of goodwill,+      work stoppage, computer failure or malfunction, or any and all+      other commercial damages or losses), even if such Contributor+      has been advised of the possibility of such damages.++   9. Accepting Warranty or Additional Liability. While redistributing+      the Work or Derivative Works thereof, You may choose to offer,+      and charge a fee for, acceptance of support, warranty, indemnity,+      or other liability obligations and/or rights consistent with this+      License. However, in accepting such obligations, You may act only+      on Your own behalf and on Your sole responsibility, not on behalf+      of any other Contributor, and only if You agree to indemnify,+      defend, and hold each Contributor harmless for any liability+      incurred by, or claims asserted against, such Contributor by reason+      of your accepting any such warranty or additional liability.++   END OF TERMS AND CONDITIONS++   APPENDIX: How to apply the Apache License to your work.++      To apply the Apache License to your work, attach the following+      boilerplate notice, with the fields enclosed by brackets "{}"+      replaced with your own identifying information. (Don't include+      the brackets!)  The text should be enclosed in the appropriate+      comment syntax for the file format. We also recommend that a+      file or class name and description of purpose be included on the+      same "printed page" as the copyright notice for easier+      identification within third-party archives.++   Copyright {yyyy} {name of copyright owner}++   Licensed under the Apache License, Version 2.0 (the "License");+   you may not use this file except in compliance with the License.+   You may obtain a copy of the License at++       http://www.apache.org/licenses/LICENSE-2.0++   Unless required by applicable law or agreed to in writing, software+   distributed under the License is distributed on an "AS IS" BASIS,+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.+   See the License for the specific language governing permissions and+   limitations under the License.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ krapsh.cabal view
@@ -0,0 +1,148 @@+name:                krapsh+version:             0.1.6.0+synopsis:            Haskell bindings for Spark Dataframes and Datasets+description:         Please see README.md+homepage:            https://github.com/krapsh/kraps-haskell+license:             Apache-2.0+license-file:        LICENSE+author:              krapsh+maintainer:          krapsh@yandex.com+copyright:           2016 Kraps-Haskell contributors+category:            Web, Big data+build-type:          Simple+-- extra-source-files:+cabal-version:       >=1.10++library+  hs-source-dirs:      src+  exposed-modules:     Spark.Core,+                       Spark.Core.Context,+                       Spark.Core.Column,+                       Spark.Core.ColumnFunctions,+                       Spark.Core.Dataset,+                       Spark.Core.Functions,+                       Spark.Core.Internal.Caching,+                       Spark.Core.Internal.ComputeDag,+                       Spark.Core.Internal.ContextInternal,+                       Spark.Core.Internal.ContextStructures,+                       Spark.Core.Internal.DAGFunctions,+                       Spark.Core.Internal.DAGStructures,+                       Spark.Core.Internal.DatasetFunctions,+                       Spark.Core.Internal.DatasetStructures,+                       Spark.Core.Internal.LocalDataFunctions,+                       Spark.Core.Internal.OpFunctions,+                       Spark.Core.Internal.OpStructures,+                       Spark.Core.Internal.Paths,+                       Spark.Core.Internal.PathsUntyped,+                       Spark.Core.Internal.Utilities,+                       Spark.Core.Internal.TypesStructures,+                       Spark.Core.Internal.TypesFunctions,+                       Spark.Core.Row,+                       Spark.Core.StructuresInternal,+                       Spark.Core.Try,+                       Spark.Core.Types+  other-modules:       Spark.Core.Internal.CachingUntyped,+                       Spark.Core.Internal.ColumnFunctions,+                       Spark.Core.Internal.AlgebraStructures,+                       Spark.Core.Internal.ColumnStructures,+                       Spark.Core.Internal.ContextInteractive,+                       Spark.Core.Internal.ContextIOInternal,+                       Spark.Core.Internal.Client,+                       Spark.Core.Internal.AggregationFunctions,+                       Spark.Core.Internal.FunctionsInternals,+                       Spark.Core.Internal.LocatedBase,+                       Spark.Core.Internal.RowGenerics,+                       Spark.Core.Internal.RowGenericsFrom,+                       Spark.Core.Internal.RowStructures,+                       Spark.Core.Internal.RowUtils,+                       Spark.Core.Internal.TypesGenerics+  build-depends:       aeson,+                       aeson-pretty,+                       base >= 4.8.1 && < 5,+                       base16-bytestring,+                       binary == 0.7.5.0,+                       bytestring,+                       containers,+                       cryptohash-sha256,+                       deepseq,+                       exceptions,+                       formatting,+                       hashable,+                       lens,+                       monad-logger,+                       QuickCheck,+                       random,+                       scientific,+                       SHA,+                       mtl,+                       text,+                       text-format,+                       transformers,+                       unordered-containers,+                       vector,+                       wreq+  ghc-options:         -Wall+--  ghc-options:         -fprof-auto --enable-library-profiling -O0 -- -ddump-deriv -O0 -rtsopts=all+  default-language:    Haskell2010+++test-suite krapsh-test+  type:                exitcode-stdio-1.0+  hs-source-dirs:      test+  main-is:             Spec.hs+  other-modules:       Spark.Core.ContextSpec,+                       Spark.Core.DatasetSpec,+                       Spark.Core.Internal.CachingSpec,+                       Spark.Core.Internal.LocalDataFunctionsSpec,+                       Spark.Core.Internal.OpFunctionsSpec,+                       Spark.Core.Internal.RowUtilsSpec,+                       Spark.Core.Internal.DAGFunctionsSpec,+                       Spark.Core.Internal.PathsSpec,+                       Spark.Core.PathSpec,+                       Spark.Core.ProjectionsSpec,+                       Spark.Core.RowToSQLSpec,+                       Spark.Core.TypesSpec,+                       Spark.Core.ColumnSpec,+                       Spark.Core.SimpleExamplesSpec+  build-depends:       aeson+                     , base+                     , bytestring+                     , containers+                     , formatting+                     , krapsh+                     , hspec == 2.*+                     , text+                     , raw-strings-qq+                     , QuickCheck+                     , vector+  ghc-options:         -fhpc -O0 -Wall+--  ghc-options:         -ddump-deriv -O0 -rtsopts=all -prof -auto-all+  default-language:    Haskell2010++test-suite krapsh-test-integration+  type:                exitcode-stdio-1.0+  hs-source-dirs:      test-integration+  main-is:             Spec.hs+  other-modules:       Spark.Core.CachingSpec,+                       Spark.Core.CollectSpec,+                       Spark.Core.IntegrationUtilities,+                       Spark.Core.SimpleAddSpec+  build-depends:       aeson+                     , base+                     , bytestring+                     , containers+                     , formatting+                     , krapsh+                     , hspec == 2.*+                     , text+                     , raw-strings-qq+                     , QuickCheck+                     , vector+  ghc-options:         -fhpc -O0 -Wall+--  ghc-options:         -ddump-deriv -O0 -rtsopts=all -prof -auto-all+  default-language:    Haskell2010+++source-repository head+  type:     git+  location: https://github.com/krapsh/kraps-haskell
+ src/Spark/Core.hs view
@@ -0,0 +1,12 @@+{-|+Module      : Spark.Core+Description : Core functions and data structures to communicate with the krapsh+              server.+Copyright   : (c) Krapsh contributors, 2016+License     : Apache-2.0+Maintainer  : krapsh@yandex.com+Stability   : experimental+Portability : POSIX++-}+module Spark.Core where
+ src/Spark/Core/Column.hs view
@@ -0,0 +1,38 @@+{-# LANGUAGE FlexibleContexts #-}++{- |+Module      : Spark.Core.Column+Description : Column types and basic operations.++Operations on columns.+-}+module Spark.Core.Column(+  -- * Types+  Column,+  DynColumn,+  -- * Extractions and collations+  asCol,+  pack1,+  pack,+  pack',+  struct,+  struct',+  castCol,+  castCol',+  colRef,+  (//),+  -- ToStaticProjectable,+  StaticColProjection,+  DynamicColProjection,+  unsafeStaticProjection,+  -- * Column functions+  colType,+  untypedCol,+  colFromObs,+  colFromObs'+  ) where++import Spark.Core.Internal.ColumnStructures+import Spark.Core.Internal.ColumnFunctions+import Spark.Core.Internal.AlgebraStructures+import Spark.Core.Internal.FunctionsInternals
+ src/Spark/Core/ColumnFunctions.hs view
@@ -0,0 +1,17 @@++{-|+Module      : Spark.Core.ColumnFunctions+Description : Column operations++The standard library of functions that operate on+data columns.+-}+module Spark.Core.ColumnFunctions(+  -- * Arithmetic operations+  (.+),+  -- * Reductions+  colSum,+) where++import Spark.Core.Internal.AlgebraStructures+import Spark.Core.Internal.AggregationFunctions
+ src/Spark/Core/Context.hs view
@@ -0,0 +1,41 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++{- | This module defines session objects that act as entry points to spark.++There are two ways to interact with Spark: using an explicit state object,+or using the default state object (interactive session).++While the interactive session is the most convenient, it should not be+used for more than quick experimentations. Any complex code should use+the SparkSession and SparkState objects.+-}+module Spark.Core.Context(+  SparkSessionConf(..),+  SparkSession,+  SparkState,+  SparkInteractiveException,+  FromSQL,+  defaultConf,+  executeCommand1,+  createSparkSessionDef,+  closeSparkSessionDef,+  exec1Def) where++import Data.Text(pack)++import Spark.Core.Internal.ContextStructures+import Spark.Core.Internal.ContextIOInternal+import Spark.Core.Internal.ContextInteractive+import Spark.Core.Internal.RowGenericsFrom(FromSQL)+++-- | The default configuration if the krapsh server is being run locally.+defaultConf :: SparkSessionConf+defaultConf =+  SparkSessionConf {+    confEndPoint = pack "http://127.0.0.1",+    confPort = 8081,+    confPollingIntervalMillis = 500,+    confRequestedSessionName = ""+  }
+ src/Spark/Core/Dataset.hs view
@@ -0,0 +1,43 @@++{- |+Module      : Spark.Core.Dataset+Description : Dataset types and basic operations.++This module describes the core data types (Dataset, DataFrame,+Observable and DynObservable), and some basic operations to relate them.+-}+module Spark.Core.Dataset(+  -- * Common data structures+  -- TODO Should it be hidden?+  ComputeNode,+  LocLocal,+  LocDistributed,+  LocUnknown,+  UntypedNode,+  -- * Distributed data structures+  Dataset,+  DataFrame,+  -- * Local data structures+  LocalData,+  LocalFrame,+  -- * Conversions+  asDF,+  asDS,+  asLocalObservable,+  -- * Relations+  parents,+  untyped,+  depends,+  logicalParents,+  -- * Attributes+  nodeLogicalParents,+  nodeLogicalDependencies,+  nodeParents,+  nodeOp,+  nodeId,+  nodeName,+  nodeType,+  ) where++import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.DatasetFunctions
+ src/Spark/Core/Functions.hs view
@@ -0,0 +1,34 @@+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.Functions(+  dataset,+  dataframe,+  constant,+  collect,+  count,+  identity,+  autocache,+  cache,+  uncache,+  (@@)+  ) where+++import Data.Aeson(toJSON)+import qualified Data.Vector as V++import Spark.Core.Dataset+import Spark.Core.Types+import Spark.Core.Row+import Spark.Core.Internal.DatasetFunctions+import Spark.Core.Internal.Utilities+import Spark.Core.Internal.LocalDataFunctions+import Spark.Core.Internal.FunctionsInternals()+import Spark.Core.Internal.OpStructures+import Spark.Core.Internal.AggregationFunctions+import Spark.Core.Internal.TypesStructures(SQLType(..))++dataset :: (ToSQL a, SQLTypeable a) => [a] -> Dataset a+dataset l = emptyDataset op tp where+  tp = buildType+  op = NodeDistributedLit (unSQLType tp) (V.fromList ((toJSON . valueToCell) <$> l))
+ src/Spark/Core/Internal/AggregationFunctions.hs view
@@ -0,0 +1,163 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE FlexibleContexts #-}++-- A number of standard aggregation functions.++module Spark.Core.Internal.AggregationFunctions where++import Data.Aeson(Value(Null))+import qualified Data.Text as T+import Formatting++import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.ColumnStructures+import Spark.Core.Internal.DatasetFunctions+import Spark.Core.Internal.RowGenerics(ToSQL)+import Spark.Core.Internal.LocalDataFunctions()+import Spark.Core.Internal.FunctionsInternals+import Spark.Core.Internal.OpStructures+import Spark.Core.Internal.Utilities(failure, HasCallStack)+import Spark.Core.Internal.TypesStructures+import Spark.Core.Types++{-| The sum of all the elements in a column.++If the data type is too small to represent the sum, the value being returned is+undefined.+-}+colSum :: forall ref a. (Num a, SQLTypeable a, ToSQL a) =>+  Column ref a -> LocalData a+colSum = applyUniAgg (_sumAgg :: UniversalAggregator a a)++{-| The number of elements in a column.++-}+-- TODO use Long for the return data type.+count :: forall a. (SQLTypeable a) => Dataset a -> LocalData Int+count ds = applyUniAgg (_countAgg2 :: UniversalAggregator a Int) (asCol ds)++{-| Collects all the elements of a column into a list.++NOTE:+This list is sorted in the canonical ordering of the data type: however the+data may be stored by Spark, the result will always be in the same order.+This is a departure from Spark, which does not guarantee an ordering on+the returned data.+-}+collect :: forall ref a. (SQLTypeable a) => Column ref a -> LocalData [a]+collect = applyUniAgg (_collectAgg :: UniversalAggregator a [a])++{-|+This is the universal aggregator: the invariant aggregator and+some extra laws to combine multiple outputs.+It is useful for combining the results over multiple passes.+A real implementation in Spark has also an inner pass.+-}+data UniversalAggregator a buff = UniversalAggregator {+  -- The result is partioning invariant+  uaInitialOuter :: Dataset a -> LocalData buff,+  -- This operation is associative and commutative+  -- The logical parents of the final observable have to be the 2 inputs+  uaMergeBuffer :: LocalData buff -> LocalData buff -> LocalData buff+}++-- | (internal)+univAggToOp :: forall a buff. (SQLTypeable a, SQLTypeable buff) =>+  UniversalAggregator a buff -> UniversalAggregatorOp+univAggToOp = univAggToOpTyped (buildType :: SQLType a) (buildType :: SQLType buff)++-- | (internal)+univAggToOpTyped :: forall a buff.+  SQLType a ->+  SQLType buff ->+  UniversalAggregator a buff ->+  UniversalAggregatorOp+univAggToOpTyped sqlta sqltm ua =+  let+    mt = unSQLType sqltm+    outer = _unsafeExtractOp $ fun1ToOpTyped sqlta (uaInitialOuter ua)+    merge = _unsafeExtractOp $ fun2ToOpTyped sqltm sqltm (uaMergeBuffer ua)+  in UniversalAggregatorOp {+    uaoMergeType = mt,+    uaoInitialOuter = outer,+    uaoMergeBuffer = merge+  }++-- | (internal)+applyUniAgg :: UniversalAggregator a b -> Column ref a -> LocalData b+applyUniAgg ua c =+  let+    ds = pack1 c+    ld1 = uaInitialOuter ua ds+    -- TODO understand how to pass this info+    -- aggop = univAggToOpTyped (nodeType ds) (nodeType ld1) ua+    -- ld = emptyLocalData (NodeUniversalAggregator aggop) (nodeType ld1)+  in ld1+++-- (internal)+simpleOp1Typed :: (IsLocality loca, IsLocality locb) =>+  SQLType b ->+  T.Text ->+  ComputeNode loca a -> ComputeNode locb b+simpleOp1Typed sqltb name =+  let so = StandardOperator {+             soName = name,+             soOutputType = unSQLType sqltb,+             soExtra = Null+           }+      no = NodeLocalOp so+  in nodeOpToFun1Typed sqltb no++-- (internal)+simpleOp1 :: forall a b loca locb. (IsLocality loca, IsLocality locb, SQLTypeable a, SQLTypeable b) =>+  T.Text ->+  ComputeNode loca a -> ComputeNode locb b+simpleOp1 = simpleOp1Typed (buildType :: SQLType b)++-- (internal)+simpleOp2 :: forall a1 a2 b loc1 loc2 locb. (SQLTypeable b, IsLocality loc1, IsLocality loc2, IsLocality locb) =>+  T.Text ->+  ComputeNode loc1 a1 -> ComputeNode loc2 a2 -> ComputeNode locb b+simpleOp2 = simpleOp2Typed (buildType :: SQLType b)++-- (internal)+simpleOp2Typed :: (IsLocality loc1, IsLocality loc2, IsLocality locb) =>+  SQLType b ->+  T.Text ->+  ComputeNode loc1 a1 -> ComputeNode loc2 a2 -> ComputeNode locb b+simpleOp2Typed sqltb name =+  let so = StandardOperator {+             soName = name,+             soOutputType = unSQLType sqltb,+             soExtra = Null+           }+      no = NodeLocalOp so+  in nodeOpToFun2Typed sqltb no++_unsafeExtractOp :: (HasCallStack) => NodeOp -> StandardOperator+_unsafeExtractOp (NodeLocalOp so) = so+_unsafeExtractOp (NodeOpaqueAggregator so) = so+_unsafeExtractOp (NodeDistributedOp so) = so+_unsafeExtractOp x = failure $ sformat ("Expected standard op, found "%shown) x++_countAgg2 :: (SQLTypeable a) => UniversalAggregator a Int+_countAgg2 = UniversalAggregator {+    uaInitialOuter = simpleOp1 "org.spark.Count",+    uaMergeBuffer = (+)+  }++_sumAgg :: forall a. (SQLTypeable a, Num a, ToSQL a) => UniversalAggregator a a+_sumAgg = UniversalAggregator {+    uaInitialOuter = simpleOp1 "org.spark.Sum",+    uaMergeBuffer = (+)+  }++_collectAgg :: forall a. SQLTypeable a => UniversalAggregator a [a]+_collectAgg =+  UniversalAggregator {+    uaInitialOuter = simpleOp1 "org.spark.Collect",+    uaMergeBuffer = simpleOp2 "org.spark.CatSorted"+  }
+ src/Spark/Core/Internal/AlgebraStructures.hs view
@@ -0,0 +1,81 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE FunctionalDependencies #-}+++module Spark.Core.Internal.AlgebraStructures where++import Spark.Core.Try+import Spark.Core.StructuresInternal+import Spark.Core.Internal.TypesStructures++-- | Algebraic structures that are common to columns and observables.+++{-| The class of static projections that are guaranteed to succeed+by using the type system.++from is the type of the dataset (which is also a typed dataset)+to is the type of the final column.+-}+data StaticColProjection from to = StaticColProjection {+  _staticProj :: (SQLType from, FieldPath, SQLType to)+}++{-| The class of projections that require some runtime introspection+to confirm that the projection is valid.+-}+data DynamicColProjection = DynamicColProjection {+  -- The start type is irrelevant.+  _dynProjTry :: DataType -> Try (FieldPath, DataType)+}++{-| The operation of extraction from a Spark object to another+object.+-}+class Projection from proj to | from proj -> to where+  _performProjection :: from -> proj -> to++{-| The projector operation.++This is the general projection operation in Spark. It lets you extract columns+from datasets or dataframes, or sub-observables form observables.++TODO(kps) put an example here.+-}+(//) :: forall from proj to. Projection from proj to => from -> proj -> to+(//) = _performProjection++data BinaryOpFun in1 in2 to = BinaryOpFun {+  bodLift1 :: in1 -> to,+  bodLift2 :: in2 -> to,+  bodOp :: to -> to -> to+}++class HomoBinaryOp2 in1 in2 to | in1 in2 -> to where+  _liftFun :: (to -> to -> to) -> BinaryOpFun in1 in2 to++_applyBinOp0 :: forall in1 in2 to. in1 -> in2 -> BinaryOpFun in1 in2 to -> to+_applyBinOp0 i1 i2 (BinaryOpFun l1 l2 bo) = bo (l1 i1) (l2 i2)++applyBinOp :: forall in1 in2 to. (HomoBinaryOp2 in1 in2 to) => (to -> to -> to) -> in1 -> in2 -> to+applyBinOp f i1 i2 =+  _applyBinOp0 i1 i2 (_liftFun f)++-- | Overloaded operator for operationts that are guaranteed to succeed.+(.+) :: (Num out, HomoBinaryOp2 a1 a2 out) => a1 -> a2 -> out+(.+) = applyBinOp (+)++(.-) :: (Num out, HomoBinaryOp2 a1 a2 out) => a1 -> a2 -> out+(.-) = applyBinOp (-)++(.*) :: (Num out, HomoBinaryOp2 a1 a2 out) => a1 -> a2 -> out+(.*) = applyBinOp (*)++-- TODO(kps) add here the rest of the Integral operations+div' :: (Integral out, HomoBinaryOp2 a1 a2 out) => a1 -> a2 -> out+div' = applyBinOp div++-- **** Fractional ****++(./) :: (Fractional out, HomoBinaryOp2 a1 a2 out) => a1 -> a2 -> out+(./) = applyBinOp (/)
+ src/Spark/Core/Internal/Caching.hs view
@@ -0,0 +1,302 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}++{- Methods related to checking caching in the graph.+-}++module Spark.Core.Internal.Caching(+  NodeCachingType(..),+  CachingFailure(..),+  CacheTry,+  CacheGraph,+  AutocacheGen(..),+  checkCaching,+  fillAutoCache+) where++import Control.Monad.Identity+import qualified Data.Set as S+import qualified Data.Map.Strict as M+import qualified Data.Vector as V+import Control.Arrow((&&&))+import Data.Foldable+import Data.Set(Set)+import Data.Maybe(mapMaybe)+import Debug.Trace(trace)+import Data.Text(Text)+import Formatting+-- import Debug.Trace++import Spark.Core.Internal.DAGFunctions+import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.Utilities+-- import Spark.Core.StructuresInternal(NodeId)++data NodeCachingType =+    -- Hinted caching. Will be fullfilled by the algorithm below.+    -- The node id is that of the node being cached+    AutocacheOp VertexId+    -- Unconditional caching+    -- The node id is that of the node being cached+  | CacheOp VertexId+    -- First one is the node id+    -- the second is the id of the matching cache or autocache node+  | UncacheOp VertexId VertexId+  | Through+  | Stop deriving (Show, Eq)++data CachingFailure = CachingFailure {+  cachingNode :: !VertexId,+  uncachingNode :: !VertexId,+  escapingNode :: !VertexId+} deriving (Show, Eq)++type CacheTry t = Either Text t++type CacheGraph v = Graph (v, NodeCachingType) StructureEdge++data AutocacheGen v = AutocacheGen {+  -- Generates an uncaching node to insert at the final location of uncaching+  deriveUncache :: Vertex v -> Vertex v,+  -- A function that given a node, generates an identity node (and a new node+  -- id) that can be inserted in place. The generated node id will be used with+  -- the identity node newly generated; the previous node will be moved around+  -- along with its identity.+  deriveIdentity :: Vertex v -> Vertex v+}++checkCaching :: (Show v) =>+  Graph v StructureEdge ->+  (v -> CacheTry NodeCachingType) ->+  CacheTry [CachingFailure]+checkCaching g fun = _cacheGraph g fun >>= _checkCaching++fillAutoCache :: (Show v) =>+  (v -> CacheTry NodeCachingType) ->+  AutocacheGen v ->+  Graph v StructureEdge ->+  -- The final graph being constructed.+  DagTry (Graph v StructureEdge)+fillAutoCache cacheFun acGen g = do+  cg <- graphMapVertices g $ \vx _ -> (const vx &&& id) <$> cacheFun vx+  acg <- _fillAutoCache acGen cg+  let acg' = graphMapVertices' fst acg+  return acg'+++-- Some internal types to guarantee more correctness+newtype AutocacheVertex v = AutocacheVertex (Vertex v)+newtype StopVertex v = StopVertex { unStopVertex :: Vertex v }+newtype IdentityVertex v = IdentityVertex (Vertex v)+data UncacheVertex v = UncacheVertex (Vertex v) VertexId deriving (Show)+newtype AnyCacheOp = AnyCacheOp { unAnyCacheOp :: VertexId } deriving (Show, Ord, Eq)++-- The result of creating a vertex+type CreateUncache v =+  Either (UncacheVertex v) (UncacheVertex v, Edge StructureEdge)++-- This performs a graph transform:+-- For each autocache node, it finds the transitive closure of Stop nodes.+-- Then it replaces the sink nodes by a layer of identity nodes and sink nodes+-- and intercalates the uncache node between the two layers, through some+-- logical dependencies.+--+-- If it does not find the closuure, it leaves these+-- autocache nodes alone and does not attempt to remove them: they will be+-- considered as being unconditional caching without checks.+--+-- Note that it works on the reverse of the graph (flow instead of dependencies)+_fillAutoCache :: forall v. (Show v) =>+  AutocacheGen v ->+  -- The graph, already annotated with caching information+  Graph (v, NodeCachingType) StructureEdge ->+  -- The final graph being constructed.+  DagTry (Graph (v, NodeCachingType) StructureEdge)+_fillAutoCache acGen cg =+  -- Find the auto nodes.+  -- Compute the closure for each of them+  -- Perform the insertion.+  -- TODO: this function is too big, split or build subfunctions.+  let+    vxMap = M.fromList ((vertexId &&& id) <$> toList (gVertices cg))+    -- TODO: mark if the result was already in the graph+    findOrCreateIdentity :: StopVertex v -> IdentityVertex v+    findOrCreateIdentity (StopVertex vx) =+      let uvx = deriveIdentity acGen vx+      in case M.lookup (vertexId uvx) vxMap of+        Just vx' -> IdentityVertex $ fst <$> vx' -- Already created+        Nothing -> IdentityVertex uvx+    acNodesAndScopes = _autoCachingCandidates cg+    -- Add the uncaching nodes+    findOrCreateUncache' (acv, l) = case _findOrCreateUncache vxMap acGen acv of+      Left x -> trace ("findOrCreateUncache: dropping autocache node " ++ show x) Nothing+      Right (ucv, ed) -> Just (ed, (acv, ucv, l))+    acWithUncache' = mapMaybe findOrCreateUncache' acNodesAndScopes+    acWithUncache = snd <$> acWithUncache'+    acEdges = fst <$> acWithUncache'+    -- Now group by stop vertex, so that each stop vertex has a list of+    -- associated cache and uncache nodes.+    -- Not sure if they may be several, but it just sounds like good practice.+    tups = myGroupBy [(vertexId (unStopVertex svx), (cvx, uvx, svx)) | (cvx, uvx, l) <- acWithUncache,+                                              svx <- l]+    -- Just in this case, it should work because of the construction above+    -- TODO: put a lot more documentation here, it is tricky code+    group ((_, uvx, svx) : t) = (svx, findOrCreateIdentity svx, uvx : [uvx' | ( _, uvx', _) <- t])+    group [] = failure "_fillAutoCache:group: empty: should not happen"+    stopsWithCachingSteps :: [(StopVertex v, IdentityVertex v, [UncacheVertex v])]+    stopsWithCachingSteps = (group . snd) <$> M.toList tups+    tups2 = [(svx, ivx, uvx) | (svx, ivx, l) <- stopsWithCachingSteps, uvx <- l]+    folder eds (svx, ivx, uvx) = _performEdgeTransform svx ivx uvx eds+    startEdges = veEdge <$> [ve | (_, v) <- M.toList (gEdges cg), ve <- V.toList v]+    edges = acEdges ++ foldl' folder startEdges tups2+    -- Gather all the vertices and edges, and remove duplicates+    startVertices = V.toList (gVertices cg)+    ucVertices = acWithUncache <&> \(_, UncacheVertex vx cacheVid, _) ->+      -- TODO: propagate the cache vertexId with UncacheVertex+      let op = UncacheOp (vertexId vx) cacheVid+      in (id &&& const op) <$> vx+    idVertices = tups2 <&> \(_, IdentityVertex vx, _) ->+      (id &&& const Stop) <$> vx+    allVertices = startVertices ++ ucVertices ++ idVertices+    -- Make a new graph+  in buildGraphFromList allVertices edges++-- TODO: should be a try to perform extra check operations+_findOrCreateUncache :: (HasCallStack, Show v) =>+  M.Map VertexId (Vertex (v, NodeCachingType)) ->+  AutocacheGen v ->+  AutocacheVertex v -> CreateUncache v+_findOrCreateUncache vxMap acGen (AutocacheVertex acv) =+  let uvx = deriveUncache acGen acv+      acVid = vertexId acv+      uVid = vertexId uvx+      look = vertexData <$> M.lookup uVid vxMap+  in case look of+    Just (x, UncacheOp _ _) ->+      -- That vertex already exists, we will not try to create+      -- an uncaching node then+      Left $ UncacheVertex (Vertex uVid x) uVid+    Just _ ->+      -- That vertex already exists, but it is not the proper type.+      -- This is a programming error in AutocacheGen: we abort here.+      failure $ sformat ("_findOrCreateUncache:"%sh%"->"%sh) acv look+    Nothing ->+      -- The uncache node does not exist, we are going to create one.+      let ed' = Edge uVid acVid ParentEdge+      in Right (UncacheVertex uvx uVid, ed')+++-- FIXME: duplicated work on the stop and identity: pass all the uncache vertexes to process them in one go+_performEdgeTransform ::+  StopVertex v -> IdentityVertex v -> UncacheVertex v -> [Edge StructureEdge] -> [Edge StructureEdge]+_performEdgeTransform (StopVertex svx) (IdentityVertex ivx) (UncacheVertex uvx _) eds =+  let stopVid = vertexId svx+      idenVid = vertexId ivx+      ucVid = vertexId uvx+      -- Rewrite the edges incoming to the stop node so that they point to the+      -- id node instead.+      f ed | edgeTo ed == stopVid = ed { edgeTo = idenVid }+      f ed = ed+      joinEd = Edge { edgeFrom = idenVid, edgeTo = stopVid, edgeData = ParentEdge }+      id1Ed = Edge { edgeFrom = idenVid, edgeTo = ucVid, edgeData = LogicalEdge }+      id2Ed = Edge { edgeFrom = ucVid, edgeTo = stopVid, edgeData = LogicalEdge }+  in id1Ed : id2Ed : joinEd : (f <$> eds)++-- The list of nodes that do autocaching, and the fringes for each of these+-- nodes.+-- Returns a list of caching node -> [stop node]+_autoCachingCandidates :: forall v. (Show v) =>+  Graph (v, NodeCachingType) StructureEdge ->+  [(AutocacheVertex v, [StopVertex v])]+_autoCachingCandidates cg =+  let+    cg' = graphMapVertices' snd cg+    exps = gVertices $ _expansions cg'+    extractAutocache vx = case snd (vertexData vx) of+      AutocacheOp _ -> [AutocacheVertex (fst <$> vx)]+      _ -> []+    acVxs = concatMap extractAutocache (gVertices cg)+    -- All the stop nodes for each caching vertex id+    extractFringe vx = case vertexData vx of+      (Stop, set) -> (id &&& const (vertexId vx)) <$> toList set+      _ -> []+    -- cache vid -> Stop vertex id+    acWithFringe = myGroupBy $ concatMap extractFringe (toList exps)+    vmap = vertexMap cg+    vmap' = vertexMap cg'+    -- TODO: should be a try and it should not fail+    findStop :: VertexId -> Maybe (StopVertex v)+    findStop vid = do+      vx <- M.lookup vid vmap+      _ <- M.lookup vid vmap'+      return $ StopVertex (Vertex vid (fst vx))+    -- TODO: it should be a try, although it is a programming error here+    combineWithFringe :: AutocacheVertex v -> (AutocacheVertex v, [StopVertex v])+    combineWithFringe acv @ (AutocacheVertex vx) =+      let vids = M.findWithDefault [] (AnyCacheOp (vertexId vx)) acWithFringe+      in (acv, mapMaybe findStop vids)+    -- Remove the nodes that do not have a fringe.+    -- In this case, they are passed through without uncaching operation.+    acWithFringeVx = filter (not.null.snd) $ combineWithFringe <$> acVxs+  in acWithFringeVx++_checkCaching :: Graph NodeCachingType StructureEdge -> CacheTry [CachingFailure]+_checkCaching cg =+  let+    expands = snd <$> vertexMap (_expansions cg)+    removals = vertexMap $ _removals cg+    f :: NodeCachingType -> [(VertexId, VertexId)]+    f (UncacheOp uncacheNid cacheNid) = [(cacheNid, uncacheNid)]+    f _ = []+    -- cacheNID -> uncacheNID+    removedNodes = M.fromList $ concatMap f (vertexData <$> gVertices cg)+    removedNodeSet = S.fromList $ M.keys removedNodes+    checkErrors :: VertexId -> [CachingFailure]+    checkErrors nid =+      let rems = S.intersection+                  removedNodeSet+                  (M.findWithDefault S.empty nid removals)+          exps = S.intersection+                   removedNodeSet+                   (unAnyCacheOp `S.map` M.findWithDefault S.empty nid expands)+          bad = S.toList $ S.difference exps rems+          badWithUncache = flip mapMaybe bad $ \ cacheNid ->+            M.lookup cacheNid removedNodes <&> \uncacheNid ->+              CachingFailure cacheNid uncacheNid nid+      in badWithUncache+  in return $ concatMap checkErrors (vertexId <$> gVertices cg)++_cacheGraph :: (Show v) => Graph v StructureEdge ->+  (v -> CacheTry NodeCachingType) ->+  CacheTry (Graph NodeCachingType StructureEdge)+_cacheGraph g f =+  graphMapVertices g f' where+    f' vx _ = f vx++-- The set of node caching operations at each step.+-- This includes both regular cache and autocache.+_expansions :: (Show e) =>+  Graph NodeCachingType e ->+  Graph (NodeCachingType, Set AnyCacheOp) e+_expansions g = runIdentity (graphMapVertices g f) where+  f x l = return (x, S.union seta parentSet) where+    filt ((Stop, _), _) = S.empty+    -- Uncaching drops the caching node from the expansions+    filt ((UncacheOp _ cacheVid, s), _) = S.delete (AnyCacheOp cacheVid) s+    filt ((_, s), _) = s+    parentSet :: S.Set AnyCacheOp+    parentSet = S.unions (filt <$> l)+    seta = case x of+      CacheOp nid -> S.singleton (AnyCacheOp nid)+      AutocacheOp nid -> S.singleton (AnyCacheOp nid)+      _ -> S.empty++_removals :: (Show e) =>+  Graph NodeCachingType e -> Graph (Set VertexId) e+_removals g = runIdentity (graphMapVertices (reverseGraph g) f) where+  f x l = return $ S.union seta (S.unions (fst <$> l)) where+    seta = case x of+      UncacheOp _ cacheNid -> S.singleton cacheNid+      _ -> S.empty
+ src/Spark/Core/Internal/CachingUntyped.hs view
@@ -0,0 +1,58 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}+++{-| Implementation of the caching interfaces for the compute data structures.+-}+module Spark.Core.Internal.CachingUntyped(+  cachingType,+  autocacheGen+) where++import Control.Monad.Except++import Spark.Core.Internal.Caching+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.DatasetFunctions+import Spark.Core.Internal.OpStructures+import Spark.Core.Internal.PathsUntyped()+import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.Utilities+import Spark.Core.StructuresInternal++cachingType :: UntypedNode -> CacheTry NodeCachingType+cachingType n = traceHint ("cachingType: n="<>show' (nodeOp n)<>" res=") $ case nodeOp n of+  NodeLocalOp _ -> pure Stop+  NodeUniversalAggregator _ -> pure Stop+  NodeOpaqueAggregator _ -> pure Stop+  NodeLocalLit _ _ -> pure Stop+  NodeStructuredTransform _ -> pure Through+  NodeDistributedLit _ _ -> pure Through+  NodeDistributedOp so | soName so == opnameCache ->+    pure $ CacheOp (vertexToId n)+  NodeDistributedOp so | soName so == opnameUnpersist ->+    case nodeParents n of+      [n'] -> pure $ UncacheOp (vertexToId n) (vertexToId n')+      _ -> throwError "Node is not valid uncache node"+  NodeDistributedOp so | soName so == opnameAutocache ->+    pure $ AutocacheOp (vertexToId n)+  NodeDistributedOp _ -> pure Through -- Nothing special for the other operations+++autocacheGen :: AutocacheGen UntypedNode+autocacheGen = AutocacheGen {+  deriveUncache = deriveUncache',+  deriveIdentity = deriveIdentity'+} where+    -- TODO: use path-based identification in the future+    -- f :: String -> VertexId -> VertexId+    -- f s (VertexId bs) = VertexId . C8.pack . (++s) . C8.unpack $ bs+    deriveIdentity' (Vertex _ un) =+      let x = identity un+          vid' = VertexId . unNodeId . nodeId $ x -- f "_identity" vid+      in Vertex vid' x+    deriveUncache' (Vertex _ un) =+      let x = uncache un+          vid' = VertexId . unNodeId . nodeId $ x -- f "_uncache" vid+      in Vertex vid' x
+ src/Spark/Core/Internal/Client.hs view
@@ -0,0 +1,114 @@+{-# LANGUAGE DeriveGeneric #-}+-- The communication protocol with the server++module Spark.Core.Internal.Client where++import Spark.Core.StructuresInternal+import Spark.Core.Dataset(UntypedNode)+import Spark.Core.Internal.Utilities+++import Data.Text(Text, pack)+import Data.Aeson+import Data.Aeson.Types(Parser)+import GHC.Generics+++-- Imports for the client+++data LocalSessionId = LocalSessionId {+  unLocalSession :: !Text+} deriving (Eq, Show)++data ComputationID = ComputationID {+  unComputationID :: !Text+} deriving (Eq, Show, Generic)++data Computation = Computation {+  cSessionId :: !LocalSessionId,+  cId :: !ComputationID,+  cNodes :: ![UntypedNode],+  -- Non-empty+  cTerminalNodes :: ![NodeName],+  -- The node at the top of the computation.+  -- Must be part of the terminal nodes.+  cCollectingNode :: NodeName+} deriving (Show, Generic)+++data PossibleNodeStatus =+    NodeQueued+  | NodeRunning+  | NodeFinishedSuccess NodeComputationSuccess+  | NodeFinishedFailure NodeComputationFailure deriving (Show, Generic)++data NodeComputationSuccess = NodeComputationSuccess {+  -- Because Row requires additional information to be deserialized.+  ncsData :: Value+} deriving (Show, Generic)++data NodeComputationFailure = NodeComputationFailure {+  ncfMessage :: !Text+} deriving (Show, Generic)++-- data EndNodeStatus = EndNodeStatus {+--   ensPath :: !NodePath,+--   ensStatus :: !PossibleNodeStatus,+--   ensValue :: !(Maybe Cell)+-- } deriving (Show, Generic)++-- data AggregateProgressStatus = AggregateProgressStatus {+--   apsBundlePaths :: ![NodePath]+-- } deriving (Show, Generic)++-- data GetNodeStatusResponse = GetNodeStatusResponse {+--   gnsrEndNodes :: ![EndNodeStatus],+--   gnsrBundles :: ![AggregateProgressStatus]+-- } deriving (Show, Generic)++-- data GetNodeStatus = GetNodeStatus {+--   gnsSessionId :: !LocalSessionId,+--   gnsComputationId :: !ComputationID,+--   gnsIncludeResults :: !Bool,+--   gnsRequestedNodes :: ![NodePath]+-- } deriving (Show, Generic)++-- **** AESON INSTANCES ***++instance ToJSON Computation++instance ToJSON LocalSessionId where+  toJSON = toJSON . unLocalSession++-- instance FromJSON LocalSessionID where+--   parseJSON = LocalSessionID <$> parseJSON++instance ToJSON ComputationID where+  toJSON = toJSON . unComputationID++-- Because we get a row back, we need to supply a SQLType for deserialization.+instance FromJSON PossibleNodeStatus where+  parseJSON =+    let parseSuccess :: Object -> Parser PossibleNodeStatus+        parseSuccess o =+          (NodeFinishedSuccess . NodeComputationSuccess) <$> o .: pack "finalResult"+        parseFailure :: Object -> Parser PossibleNodeStatus+        parseFailure o =+          (NodeFinishedFailure . NodeComputationFailure) <$> o .: pack "finalError"+    in+      withObject "PossibleNodeStatus" $ \o -> do+      status <- o .: pack "status"+      case status of+        "running" -> return NodeRunning+        "finished_success" -> parseSuccess o+        "finished_failure" -> parseFailure o+        "scheduled" -> return NodeQueued+        _ -> failure $ pack ("FromJSON PossibleNodeStatus " ++ show status)++    -- Person <$> o .: "name" <*> o .: "age"++-- instance ToJSON PossibleNodeStatus where+--   toJSON NodeQueued = toJSON "queued"+--   toJSON NodeRunning = toJSON "running"+--   toJSON NodeFinished = toJSON "finished"
+ src/Spark/Core/Internal/ColumnFunctions.hs view
@@ -0,0 +1,409 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}++-- Implements a DSL for extracting some columns from dataframes and datasets.++module Spark.Core.Internal.ColumnFunctions(+  -- Accessors+  colType,+  colOrigin,+  colOp,+  colFieldName,+  -- Internal API+  iUntypedColData,+  iEmptyCol,+  -- Developer API (projections)+  unsafeStaticProjection,+  dynamicProjection,+  -- Public functions+  untypedCol,+  colFromObs,+  colFromObs',+  castCol,+  castCol',+  colRef+) where++import qualified Data.Text as T+import qualified Data.Text.Format as TF+import qualified Data.Vector as V+import Data.String(IsString(fromString))+import Data.Text.Lazy(toStrict)+import Data.Maybe(fromMaybe)+import Data.List(find)+import Formatting+++import Spark.Core.Internal.ColumnStructures+import Spark.Core.Internal.DatasetFunctions+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.TypesStructures+import Spark.Core.Try+import Spark.Core.StructuresInternal+import Spark.Core.Internal.TypesFunctions+import Spark.Core.Internal.OpStructures+import Spark.Core.Internal.OpFunctions(prettyShowColOp)+import Spark.Core.Internal.AlgebraStructures+import Spark.Core.Internal.Utilities++-- ********** Public methods ********++{-| Lets the users define their own static projections.++Throws an error if the type cannot be found, so should be used with caution.++String has to be used because of type inferrence issues+-}+unsafeStaticProjection :: forall from to. (HasCallStack) =>+  SQLType from     -- ^ The start type+  -> String        -- ^ The name of a field assumed to be found in the start type.+                   --   This only has to be valid for Spark purposes, not+                   --   internal Haskell representation.+  -> StaticColProjection from to+unsafeStaticProjection sqlt field =+  let+    f = forceRight . fieldPath . T.pack $ field+    sqlt' = fromMaybe+      (failure $ sformat ("unsafeStaticProjection: Cannot find the field "%sh%" in type "%sh) field sqlt)+      (_extractPath sqlt f)+  in StaticColProjection (sqlt, f, sqlt')++-- Returns a projection from a path (even if invalid data)+dynamicProjection :: String -> DynamicColProjection+dynamicProjection txt = case fieldPath (T.pack txt) of+  Left msg -> DynamicColProjection $ \_ ->+    tryError $ sformat ("dynamicProjection: invalid syntax for path "%shown%": "%shown) txt msg+  Right fpath -> _dynamicProjection fpath++{-| The type of a column.+-}+colType :: Column ref a -> SQLType a+colType = SQLType . _cType++{-| Converts a type column to an antyped column.+-}+untypedCol :: Column ref a -> DynColumn+untypedCol = pure . _unsafeCastColData . _dropReference++{-| Casts a dynamic column to a statically typed column.++In this case, one must supply the reference (which can be obtained from+another column with colRef, or from a dataset), and a type (which can be+built using the buildType function).+-}+castCol :: ColumnReference ref -> SQLType a -> DynColumn -> Try (Column ref a)+castCol r sqlt dc =+  dc >>= _checkedCastColData sqlt >>= _checkedCastRefColData r++{-| Casts a dynamic column to a statically typed column, but does not attempt+to enforce a single origin at the type level.++This is useful when building a dataset from a dataframe: the origin information+cannot be conveyed since it is not available in the first place.+-}+castCol' :: SQLType a -> DynColumn -> Try (Column UnknownReference a)+castCol' = castCol ColumnReference++-- (internal)+colOrigin :: Column ref a -> UntypedDataset+colOrigin = _cOrigin++-- (internal)+colOp :: Column ref a -> ColOp+colOp = _cOp++{-| A tag with the reference of a column.++This is useful when casting dynamic columns to typed columns.+-}+colRef :: Column ref a -> ColumnReference ref+colRef _ = ColumnReference++-- | Takes an observable and makes it available as a column of the same type.+colFromObs :: (HasCallStack) => LocalData a -> Column (LocalData a) a+colFromObs = missing "colFromObs"++-- | Takes a dynamic observable and makes it available as a dynamic column.+colFromObs' :: (HasCallStack) => LocalFrame -> DynColumn+colFromObs' = missing "colFromObs'"++-- | (internal)+colFieldName :: ColumnData ref a -> FieldName+colFieldName c =+  fromMaybe (unsafeFieldName . prettyShowColOp . _cOp $ c)+    (_cReferingPath c)++-- ********* Projection operations ***********++-- dataset -> static projection -> column+instance forall a to. Projection (Dataset a) (StaticColProjection a to) (Column a to) where+  _performProjection = _project++-- dataset -> dynamic projection -> DynColumn+instance forall a. Projection (Dataset a) DynamicColProjection DynColumn where+  _performProjection = _projectDyn++-- dataset -> string -> DynColumn+instance forall a . Projection (Dataset a) String DynColumn where+  _performProjection ds s = _projectDyn ds (_strToDynProj s)++-- dataframe -> dynamic projection -> dyncolumn+instance Projection DataFrame DynamicColProjection DynColumn where+  _performProjection = _projectDF++-- dataframe -> static projection -> dyncolumn+-- This is a relaxation as we could return Try (Column to) intead.+-- It makes more sense from an API perspective to just return a dynamic result.+instance forall a to. Projection DataFrame (StaticColProjection a to) DynColumn where+  _performProjection df proj = _projectDF df (_colStaticProjToDynProj proj)++-- dataframe -> string -> dyncolumn+instance Projection DataFrame String DynColumn where+  _performProjection df s = _projectDF df (_strToDynProj s)++-- column -> static projection -> column+instance forall ref a to. Projection (Column ref a) (StaticColProjection a to) (Column ref to) where+  _performProjection = _projectCol+++-- dyncolumn -> dynamic projection -> dyncolumn+instance Projection DynColumn DynamicColProjection DynColumn where+  _performProjection = _projectDynCol++instance forall a to. Projection DynColumn (StaticColProjection a to) DynColumn where+  _performProjection dc proj = _projectDynCol dc (_colStaticProjToDynProj proj)++class StringStuff a where+  stuffAsString :: a -> String++instance StringStuff String where+  stuffAsString = undefined++-- dyncolumn -> string -> dyncolumn+instance Projection DynColumn String DynColumn where+  _performProjection dc s = _performProjection dc (_strToDynProj s)++_strToDynProj :: String -> DynamicColProjection+_strToDynProj s =+  let+    fun dt =+      case fieldPath (T.pack s) of+        Right fp -> _dynProjTry (_dynamicProjection fp) dt+        Left msg -> tryError (T.pack msg)+  in DynamicColProjection fun++-- | Converts a static project to a dynamic projector.+_colStaticProjToDynProj :: forall from to. StaticColProjection from to -> DynamicColProjection+_colStaticProjToDynProj (StaticColProjection (SQLType dtFrom, fp, SQLType dtTo)) =+  DynamicColProjection $ \dt ->+    -- TODO factorize this as a projection on types.+    if dt /= dtFrom then+      tryError $ sformat ("Cannot convert type "%shown%" into type "%shown) dt dtFrom+    else pure (fp, dtTo)++iUntypedColData :: Column ref a -> UntypedColumnData+iUntypedColData = _unsafeCastColData . _dropReference++-- Recasts the column, trusting the user knows that the type is going to be compatible.+_unsafeCastColData :: Column ref a -> Column ref b+_unsafeCastColData c = c { _cType = _cType c }++_checkedCastColData :: SQLType b -> ColumnData ref a -> Try (ColumnData ref b)+_checkedCastColData sqlt cd =+  if (unSQLType sqlt) == (unSQLType (colType cd))+    then pure (_unsafeCastColData cd)+    else tryError $ sformat ("Cannot cast column "%sh%" to type "%sh) cd sqlt++_checkedCastRefColData :: ColumnReference ref2 -> ColumnData ref a -> Try (ColumnData ref2 a)+_checkedCastRefColData _ cd =+  -- TODO: do some dynamic checks on the origin.+  pure $ cd { _cType = _cType cd }++_dynamicProjection :: FieldPath -> DynamicColProjection+_dynamicProjection fpath =+  let+    fun dt = case _extractPath (SQLType dt) fpath of+        Just (SQLType dt') -> pure (fpath, dt') -- TODO(kps) I have a doubt+        Nothing ->+          tryError $ sformat ("unsafeStaticProjection: Cannot find the field "%shown%" in type "%shown) fpath dt+   in DynamicColProjection fun++-- TODO: take a compute node instead+_projectDyn :: Dataset from -> DynamicColProjection -> DynColumn+_projectDyn ds proj = do+  (p, dt) <- _dynProjTry proj (unSQLType . nodeType $ ds)+  _emptyDynCol ds dt p++_projectDF :: DataFrame -> DynamicColProjection -> DynColumn+_projectDF df proj = do+  node <- df+  _projectDyn node proj++_project :: Dataset from -> StaticColProjection from to -> Column from to+_project ds proj = let (_, p, sqlt) = _staticProj proj in+  iEmptyCol ds sqlt p++_projectCol :: Column ref from -> StaticColProjection from to -> Column ref to+_projectCol c (StaticColProjection (_, fp, SQLType dt)) =+  _projectColData0 c fp dt++-- Performs the data projection. This is unsafe, it does not check that the+-- field path is valid in this case, nor that the final type is valid either.+_projectColData0 :: ColumnData ref a -> FieldPath -> DataType -> ColumnData ref b+_projectColData0 cd (FieldPath p) dtTo =+  -- If the column is already a projection, flatten it.+  case colOp cd of+    -- No previous parent on an extraction -> we can safely append to that one.+    ColExtraction (FieldPath p') ->+      cd { _cOp = ColExtraction (FieldPath (p V.++ p')),+           _cType = dtTo}+    _ ->+      -- Extract from the previous column.+      cd { _cOp = ColExtraction (FieldPath p),+          _cType = dtTo}++_projectDynColData :: ColumnData ref a -> DynamicColProjection -> DynColumn+_projectDynColData cd proj =+  _dynProjTry proj (_cType cd) <&> uncurry (_projectColData0 . _dropReference $ cd)++_projectDynCol :: DynColumn -> DynamicColProjection -> DynColumn+_projectDynCol c proj = do+  cd <- c+  _projectDynColData cd proj++_extractPath :: SQLType from -> FieldPath -> Maybe (SQLType to)+_extractPath sqlt (FieldPath v) = _extractPath0 sqlt (V.toList v)++_extractPath0 :: SQLType from -> [FieldName] -> Maybe (SQLType to)+_extractPath0 sqlt [] = Just (unsafeCastType sqlt)+_extractPath0 sqlt (field : l) = do+  inner <- _extractField sqlt field+  _extractPath0 inner l++_extractField :: SQLType from -> FieldName -> Maybe (SQLType to)+_extractField (SQLType (StrictType (Struct (StructType fields)))) f =+  -- There is probably a way to make it shorter...+  let z = find (\x -> structFieldName x == f) fields in+  SQLType . structFieldType <$> z+_extractField (SQLType (NullableType (Struct (StructType fields)))) f =+  -- There is probably a way to make it shorter...+  let z = find (\x -> structFieldName x == f) fields in+  SQLType . structFieldType <$> z+_extractField _ _ = Nothing++_dropReference :: ColumnData ref a -> ColumnData UnknownReference a+_dropReference c = c {_cOp = _cOp c}++-- | (internal) creates a new column with some empty data+iEmptyCol :: Dataset a -> SQLType b -> FieldPath -> Column a b+iEmptyCol = _emptyColData++-- | (internal) Creates a new column with a dynamic type.+_emptyDynCol :: Dataset a -> DataType -> FieldPath -> DynColumn+_emptyDynCol ds dt fp = Right $ _dropReference $ _emptyColData ds (SQLType dt) fp++-- A new column data structure.+_emptyColData :: Dataset a -> SQLType b -> FieldPath -> ColumnData a b+_emptyColData ds sqlt path = ColumnData {+  _cOrigin = untypedDataset ds,+  _cType = unSQLType sqlt,+  _cOp = ColExtraction path,+  _cReferingPath = Nothing+}++-- Homogeneous operation betweet 2 columns.+_homoColOp2 :: T.Text -> Column ref x -> Column ref x -> Column ref x+_homoColOp2 opName c1 c2 =+  let co = ColFunction opName (V.fromList (colOp <$> [c1, c2]))+  in ColumnData {+      _cOrigin = _cOrigin c1,+      _cType = _cType c1,+      _cOp = co,+      _cReferingPath = Nothing }++_homoColOp2' :: T.Text -> DynColumn -> DynColumn -> DynColumn+_homoColOp2' opName c1' c2' = do+  c1 <- c1'+  c2 <- c2'+  -- TODO check same origin+  return $ _homoColOp2 opName c1 c2++-- ******** Displaying and pretty printing ************++instance forall ref a. Show (Column ref a) where+  show c =+    let+      name = case _cReferingPath c of+        Just fn -> show' fn+        Nothing -> prettyShowColOp . colOp $ c+      txt = fromString "{}{{}}->{}" :: TF.Format+      -- path = T.pack . show . _cReferingPath $ c+      -- no = prettyShowColOp . colOp $ c+      fields = T.pack . show . colType $ c+      nn = unNodeName . nodeName . _cOrigin $ c+    in T.unpack $ toStrict $ TF.format txt (name, fields, nn)++-- *********** Renaming instances **************++instance forall ref a. CanRename (ColumnData ref a) FieldName where+  c @@ fn = c { _cReferingPath = Just fn }+++instance forall ref a. CanRename (Column ref a) String where+  c @@ str = case fieldName (T.pack str) of+    Right fp -> c @@ fp+    Left msg ->+      -- The syntax check here is pretty lenient, so it fails, it has+      -- some good reasons. We stop here.+      failure $ sformat ("Could not make a field path out of string "%shown%" for column "%shown%":"%shown) str c msg++instance CanRename DynColumn FieldName where+  (Right cd) @@ fn = Right (cd @@ fn)+  -- TODO better error handling+  x @@ _ = x++instance CanRename DynColumn String where+  -- An error could happen when making a path out of a string.+  (Right cd) @@ str = case fieldName (T.pack str) of+    Right fp -> Right $ cd @@ fp+    Left msg ->+      -- The syntax check here is pretty lenient, so it fails, it has+      -- some good reasons. We stop here.+      tryError $ sformat ("Could not make a field path out of string "%shown%" for column "%shown%":"%shown) str cd msg+  -- TODO better error handling+  x @@ _ = x++-- *********** Arithmetic operations **********+++instance forall a. HomoBinaryOp2 a a a where+  _liftFun f = BinaryOpFun id id f++instance forall ref a. HomoBinaryOp2 (Column ref a) DynColumn DynColumn where+  _liftFun f = BinaryOpFun untypedCol id f++instance forall ref a. HomoBinaryOp2 DynColumn (Column ref a) DynColumn where+  _liftFun f = BinaryOpFun id untypedCol f+++instance (Num x) => Num (Column ref x) where+  (+) = _homoColOp2 "sum"+  (*) _ _ = missing "Num (Column x): *"+  abs _ = missing "Num (Column x): abs"+  signum _ = missing "Num (Column x): signum"+  fromInteger _ = missing "Num (Column x): fromInteger"+  negate _ = missing "Num (Column x): negate"++instance Num DynColumn where+  (+) = _homoColOp2' "sum"+  (*) _ _ = missing "Num (DynColumn x): *"+  abs _ = missing "Num (DynColumn x): abs"+  signum _ = missing "Num (DynColumn x): signum"+  fromInteger _ = missing "Num (DynColumn x): fromInteger"+  negate _ = missing "Num (DynColumn x): negate"
+ src/Spark/Core/Internal/ColumnStructures.hs view
@@ -0,0 +1,71 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Spark.Core.Internal.ColumnStructures where++import Control.Arrow ((&&&))+import Data.Function(on)++import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.RowStructures+import Spark.Core.Internal.TypesStructures+import Spark.Core.Internal.OpStructures+import Spark.Core.StructuresInternal+import Spark.Core.Try++{-| The data structure that implements the notion of data columns.++The type on this one may either be a Cell or a proper type.++A column of data from a dataset+The ref is a reference potentially to the originating+dataset, but it may be more general than that to perform+type-safe tricks.+TODO: try something like this https://www.vidarholen.net/contents/junk/catbag.html+-}+data ColumnData ref a = ColumnData {+  -- TODO replace by a maybe to have an easier way to deal with constant+  -- columns.+  _cOrigin :: !UntypedDataset,+  _cType :: !DataType,+  _cOp :: !ColOp,+  -- The name in the dataset.+  -- If not set, it will be deduced from the operation.+  _cReferingPath :: !(Maybe FieldName)+}++{-| A column of data from a dataset or a dataframe.++This column is typed: the operations on this column will be+validdated by Haskell' type inferenc.+-}+type Column ref a = ColumnData ref a++{-| An untyped column of data from a dataset or a dataframe.++This column is untyped and may not be properly constructed. Any error+will be found during the analysis phase at runtime.+-}+type DynColumn = Try (ColumnData UnknownReference Cell)+++-- | (internal)+-- The type of untyped column data.+type UntypedColumnData = ColumnData UnknownReference Cell++{-| A dummy data type that indicates the data referenc is missing.+-}+data UnknownReference++{-| A tag that carries the reference information of a column at a+type level. This is useful when creating column.++See ref and colRef.+-}+data ColumnReference a = ColumnReference++instance forall ref a. Eq (ColumnData ref a) where+  (==) = (==) `on` (_cOrigin &&& _cType &&& _cOp &&& _cReferingPath)
+ src/Spark/Core/Internal/ComputeDag.hs view
@@ -0,0 +1,80 @@+++module Spark.Core.Internal.ComputeDag where++import Data.Foldable(toList)+import qualified Data.Map.Strict as M+import qualified Data.Vector as V+import Data.Vector(Vector)++import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.DAGFunctions++{-| A DAG of computation nodes.++At a high level, it is a total function with a number of inputs and a number+of outputs.++Note about the edges: the edges flow along the path of dependencies.++-}+data ComputeDag v e = ComputeDag {+  -- The edges that make up the DAG+  cdEdges :: !(AdjacencyMap v e),+  -- All the vertices of the graph+  -- Sorted by lexicographic order + node id for uniqueness+  cdVertices :: !(Vector (Vertex v)),+  -- The inputs of the computation graph. These correspond to the+  -- sinks of the dependency graph.+  cdInputs :: !(Vector (Vertex v)),+  -- The outputs of the computation graph. These correspond to the+  -- sources of the dependency graph.+  cdOutputs :: !(Vector (Vertex v))+} deriving (Show)+++-- | Conversion+computeGraphToGraph :: ComputeDag v e -> Graph v e+computeGraphToGraph cg =+  Graph (cdEdges cg) (cdVertices cg)++-- | Conversion+graphToComputeGraph :: Graph v e -> ComputeDag v e+graphToComputeGraph g =+  ComputeDag {+    cdEdges = gEdges g,+    cdVertices = gVertices g,+    -- We work on the graph of dependencies (not flows)+    -- The sources correspond to the outputs.+    cdInputs = V.fromList $ graphSinks g,+    cdOutputs = V.fromList $ graphSources g+  }++_mapVerticesAdj :: (Vertex v -> v') -> AdjacencyMap v e -> AdjacencyMap v' e+_mapVerticesAdj f m =+  let f1 ve =+        let vx = veEndVertex ve+            d' = f vx in+          ve { veEndVertex = vx { vertexData = d' } }+      f' v = f1 <$> v+  in M.map f' m++mapVertices :: (Vertex v -> v') -> ComputeDag v e -> ComputeDag v' e+mapVertices f cd =+  let f' vx = vx { vertexData = f vx }+  in ComputeDag {+      cdEdges = _mapVerticesAdj f (cdEdges cd),+      cdVertices = f' <$> cdVertices cd,+      cdInputs = f' <$> cdInputs cd,+      cdOutputs = f' <$> cdOutputs cd+    }++mapVertexData :: (v -> v') -> ComputeDag v e -> ComputeDag v' e+mapVertexData f = mapVertices (f . vertexData)++buildCGraph :: (GraphOperations v e, Show v, Show e) =>+  v -> DagTry (ComputeDag v e)+buildCGraph n = graphToComputeGraph <$> buildGraph n++graphDataLexico :: ComputeDag v e -> [v]+graphDataLexico cd = vertexData <$> toList (cdVertices cd)
+ src/Spark/Core/Internal/ContextIOInternal.hs view
@@ -0,0 +1,223 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}++module Spark.Core.Internal.ContextIOInternal(+  returnPure,+  createSparkSession,+  createSparkSession',+  executeCommand1+) where++import Control.Concurrent(threadDelay)+import Control.Lens((^.))+import Control.Monad.State(mapStateT, get)+import Control.Monad(forM)+import Data.Aeson(toJSON)+import Data.Functor.Identity(runIdentity)+import Data.Text(Text, pack)+import qualified Data.Text as T+import qualified Network.Wreq as W+import Network.Wreq(responseBody)+import Control.Monad.Trans(lift)+import Control.Monad.Logger(runStdoutLoggingT, LoggingT, logDebugN, logInfoN, MonadLoggerIO)+import System.Random(randomIO)+import Data.Word(Word8)+import Control.Monad.IO.Class+-- import Formatting+import Network.Wreq.Types(Postable)+import Data.ByteString.Lazy(ByteString)++import Spark.Core.Dataset+import Spark.Core.Internal.Client+import Spark.Core.Internal.ContextInternal+import Spark.Core.Internal.ContextStructures+import Spark.Core.Row+import Spark.Core.StructuresInternal+import Spark.Core.Try+import Spark.Core.Internal.Utilities++++returnPure :: forall a. SparkStatePure a -> SparkState a+returnPure p = lift $ mapStateT (return . runIdentity) p++{- | Creates a new Spark session.++This session is unique, and it will not try to reconnect to an existing+session.+-}+createSparkSession :: (MonadLoggerIO m) => SparkSessionConf -> m SparkSession+createSparkSession conf = do+  sessionName <- case confRequestedSessionName conf of+    "" -> liftIO _randomSessionName+    x -> pure x+  let session = _createSparkSession conf sessionName 0+  let url = _sessionEndPoint session+  logDebugN $ "Creating spark session at url: " <> url+  -- TODO get the current counter from remote+  _ <- _ensureSession session+  return session++{-| Convenience function for simple cases that do not require monad stacks.+-}+createSparkSession' :: SparkSessionConf -> IO SparkSession+createSparkSession' = _runLogger . createSparkSession++{- |+Executes a command:+- performs the transforms and the optimizations in the pure state+- sends the computation to the backend+- waits for the terminal nodes to reach a final state+- commits the final results to the state++If any failure is detected that is internal to Krapsh, it returns an error.+If the error comes from an underlying library (http stack, programming failure),+an exception may be thrown instead.+-}+executeCommand1 :: forall a. (FromSQL a, HasCallStack) =>+  LocalData a -> SparkState (Try a)+executeCommand1 ld = do+    session <- get+    tcomp <- returnPure $ prepareExecution1 ld+    case tcomp of+      Left err ->+        return (Left err)+      Right comp ->+        let+          obss = getTargetNodes comp+          fun3 ld2 = do+            result <- _waitSingleComputation session comp (nodeName ld2)+            return (ld2, result)+          nodeResults :: SparkState [(LocalData Cell, FinalResult)]+          nodeResults = sequence (fun3 <$> obss)+        in do+          _ <- _sendComputation session comp+          nrs <- nodeResults+          tcell <- returnPure $ storeResults comp nrs+          return $ tcell >>= (tryEither . cellToValue)++_randomSessionName :: IO Text+_randomSessionName = do+  ws <- forM [1..10] (\(_::Int) -> randomIO :: IO Word8)+  let ints = (`mod` 10) <$> ws+  return . T.pack $ "session" ++ concat (show <$> ints)++type DefLogger a = LoggingT IO a++_runLogger :: DefLogger a -> IO a+_runLogger = runStdoutLoggingT++_post :: (MonadIO m, Postable a) =>+  Text -> a -> m (W.Response ByteString)+_post url = liftIO . W.post (T.unpack url)++_get :: (MonadIO m) =>+  Text -> m (W.Response ByteString)+_get url = liftIO $ W.get (T.unpack url)++-- TODO move to more general utilities+-- Performs repeated polling until the result can be converted+-- to a certain other type.+-- Int controls the delay in milliseconds between each poll.+_pollMonad :: (MonadIO m) => m a -> Int -> (a -> Maybe b) -> m b+_pollMonad rec delayMillis check = do+  curr <- rec+  case check curr of+    Just res -> return res+    Nothing -> do+      _ <- liftIO $ threadDelay (delayMillis * 1000)+      _pollMonad rec delayMillis check+++-- Creates a new session from a string containing a session ID.+_createSparkSession :: SparkSessionConf -> Text -> Integer -> SparkSession+_createSparkSession conf sessionId =+  SparkSession conf sid where+    sid = LocalSessionId sessionId++-- The URL of the end point+_sessionEndPoint :: SparkSession -> Text+_sessionEndPoint sess =+  let port = (pack . show . confPort . ssConf) sess+      sid = (unLocalSession . ssId) sess+  in+    T.concat [+      (confEndPoint . ssConf) sess, ":", port,+      "/session/", sid]++_sessionPortText :: SparkSession -> Text+_sessionPortText = pack . show . confPort . ssConf++-- The URL of the computation end point+_compEndPoint :: SparkSession -> ComputationID -> Text+_compEndPoint sess compId =+  let port = _sessionPortText sess+      sid = (unLocalSession . ssId) sess+      cid = unComputationID compId+  in+    T.concat [+      (confEndPoint . ssConf) sess, ":", port,+      "/computation/", sid, "/", cid]++-- The URL of the status of a computation+_compEndPointStatus :: SparkSession -> ComputationID -> Text+_compEndPointStatus sess compId =+  let port = _sessionPortText sess+      sid = (unLocalSession . ssId) sess+      cid = unComputationID compId+  in+    T.concat [+      (confEndPoint . ssConf) sess, ":", port,+      "/status/", sid, "/", cid]++-- Ensures that the server has instantiated a session with the given ID.+_ensureSession :: (MonadLoggerIO m) => SparkSession -> m ()+_ensureSession session = do+  let url = _sessionEndPoint session <> "/create"+  -- logDebugN $ "url:" <> url+  _ <- _post url (toJSON 'a')+  return ()+++_sendComputation :: (MonadLoggerIO m) => SparkSession -> Computation -> m ()+_sendComputation session comp = do+  let base' = _compEndPoint session (cId comp)+  let url = base' <> "/create"+  logInfoN $ "Sending computations at url: " <> url+  _ <- _post url (toJSON (cNodes comp))+  return ()++_computationStatus :: (MonadLoggerIO m) =>+  SparkSession -> ComputationID -> NodeName -> m PossibleNodeStatus+_computationStatus session compId nname = do+  let base' = _compEndPointStatus session compId+  let rest = unNodeName nname+  let url = base' <> "/" <> rest+  logDebugN $ "Sending computations status request at url: " <> url+  _ <- _get url+  -- raw <- _get url+  --logDebugN $ sformat ("Got raw status: "%sh) raw+  status <- liftIO (W.asJSON =<< W.get (T.unpack url) :: IO (W.Response PossibleNodeStatus))+  --logDebugN $ sformat ("Got status: "%sh) status+  let s = status ^. responseBody+  case s of+    NodeFinishedSuccess _ -> logInfoN $ rest <> " finished: success"+    NodeFinishedFailure _ -> logInfoN $ rest <> " finished: failure"+    _ -> return ()+  return s+++_waitSingleComputation :: (MonadLoggerIO m) =>+  SparkSession -> Computation -> NodeName -> m FinalResult+_waitSingleComputation session comp nname =+  let+    extract :: PossibleNodeStatus -> Maybe FinalResult+    extract (NodeFinishedSuccess s) = Just $ Right s+    extract (NodeFinishedFailure f) = Just $ Left f+    extract _ = Nothing+    -- getStatus :: m PossibleNodeStatus+    getStatus = _computationStatus session (cId comp) nname+    i = confPollingIntervalMillis $ ssConf session+  in+    _pollMonad getStatus i extract
+ src/Spark/Core/Internal/ContextInteractive.hs view
@@ -0,0 +1,123 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}++-- | Functions to create and manipulate one default context.+--+-- This is most appropriate when working in an interactive session,+-- during which it is usually clear that there is a single+-- Spark context in use.+--+-- This module uses unsafe Haskell code that should not be used+-- outside prototyping in an interactive REPL. In any good case,+-- you should use the SparkState monad.+module Spark.Core.Internal.ContextInteractive(+  SparkInteractiveException,+  createSparkSessionDef,+  exec1Def,+  closeSparkSessionDef+) where++import qualified Data.Vector as V+import Control.Exception+import Control.Monad.Catch(throwM)+import Data.IORef+import Data.Typeable+import Control.Monad.State(runStateT)+import Data.Text+import System.IO.Unsafe(unsafePerformIO)+import Control.Monad.Logger(runStdoutLoggingT)+++import Spark.Core.Internal.ContextStructures+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.ContextIOInternal+import Spark.Core.Internal.RowGenericsFrom(FromSQL)+import Spark.Core.Internal.Utilities+import Spark.Core.StructuresInternal+import Spark.Core.Try++-- The global session reference. Should not be accessed outside+-- this file.+_globalSessionRef :: IORef (Maybe SparkSession)+{-# NOINLINE _globalSessionRef #-}+_globalSessionRef = unsafePerformIO (newIORef Nothing)++-- | The exception thrown when a request cannot be completed+-- in an interactive session.+data SparkInteractiveException = SparkInteractiveException {+  _sieInner :: NodeError+} deriving Typeable++instance Show SparkInteractiveException where+  show (SparkInteractiveException inner) =+    show inner++instance Exception SparkInteractiveException++{- | Creates a spark session that will be used as the default session.++If a session already exists, an exception will be thrown.+ -}+createSparkSessionDef :: (HasCallStack) => SparkSessionConf -> IO ()+createSparkSessionDef conf = do+  current <- _currentSession+  case current of+    Nothing ->+      return ()+    Just _ ->+      -- TODO let users change the state+      _throw "A default context already exist. If you wish to modify the exsting context, you must use modifySparkConfDef"+  new <- createSparkSession' conf+  _setSession new+  return ()++{- | Executes a command using the default spark session.++This is the most unsafe way of running a command:+it executes a command using the default spark session, and+throws an exception if any error happens.+ -}+exec1Def :: (FromSQL a, HasCallStack) => LocalData a -> IO a+exec1Def ld = do+  mCtx <- _currentSession+  case mCtx of+    Nothing ->+      _throw "No default context found. You must first create a default spark context with createSparkSessionDef"+    Just ctx -> do+      (res, newSt) <- (runStateT . runStdoutLoggingT . executeCommand1) ld ctx+      _setSession newSt+      case res of+        Right x ->+          return x+        Left err ->+          throwM (SparkInteractiveException err)++{-| Closes the default session. The default session is empty after this call+completes.++NOTE: This does not currently clear up the resources! It is a stub implementation+used in testing.+-}+closeSparkSessionDef :: (HasCallStack) => IO ()+closeSparkSessionDef = do+  _ <- _removeSession+  return ()++_currentSession :: (HasCallStack) => IO (Maybe SparkSession)+_currentSession = readIORef _globalSessionRef++_setSession :: (HasCallStack) => SparkSession -> IO ()+_setSession st = writeIORef _globalSessionRef (Just st)++_removeSession :: (HasCallStack) => IO (Maybe SparkSession)+_removeSession = do+  current <- _currentSession+  _ <- writeIORef _globalSessionRef Nothing+  return current++_throw :: (HasCallStack) => Text -> IO a+_throw txt = throwM $+  SparkInteractiveException Error {+    ePath = NodePath V.empty,+    eMessage = txt+  }
+ src/Spark/Core/Internal/ContextInternal.hs view
@@ -0,0 +1,199 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}++-- Functions to build the graph of computations.+-- The following steps are performed:+--  - typing checking+--  - caching checks+--  - building the final json+--+-- All the functions in this module are pure and use SparkStatePure for transforms.++module Spark.Core.Internal.ContextInternal(+  FinalResult,+  prepareExecution1,+  buildComputationGraph,+  performGraphTransforms,+  getTargetNodes,+  storeResults,+) where++import Control.Monad.State(get, put)+import Control.Monad(forM)+import Data.Text(pack)+import Debug.Trace(trace)+import Data.Foldable(toList)+import Control.Arrow((&&&))+import Formatting+import qualified Data.Map.Strict as M+import qualified Data.Vector as V++import Spark.Core.Dataset+import Spark.Core.Try+import Spark.Core.Row+import Spark.Core.Types+import Spark.Core.Internal.Caching+import Spark.Core.Internal.CachingUntyped+import Spark.Core.Internal.ContextStructures+import Spark.Core.Internal.Client+import Spark.Core.Internal.ComputeDag+import Spark.Core.Internal.PathsUntyped+-- Required to import the instances.+import Spark.Core.Internal.Paths()+import Spark.Core.Internal.TypesStructures+import Spark.Core.Internal.TypesFunctions(arrayType)+import Spark.Core.Internal.DAGFunctions(buildVertexList)+import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.DatasetFunctions+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.Utilities++-- The result from querying the status of a computation+type FinalResult = Either NodeComputationFailure NodeComputationSuccess+++-- The main function that takes a single output point and+-- tries to transform it as a valid computation.+prepareExecution1 :: LocalData a -> SparkStatePure (Try Computation)+prepareExecution1 ld = get >>= \session ->+  let cg = buildComputationGraph ld+      cg' = performGraphTransforms =<< cg+      comp = _buildComputation session =<< cg'+  in case comp of+      Left _ -> return comp+      Right _ -> do+        _increaseCompCounter+        return comp++-- Here are the steps being run+--  - node collection + cycle detection+--  - naming:+--    -> everything after that can be done with names, and on server+--    -> for convenience, the vertex ids will be still the hash ids+--  - verification of cache/uncache+--  - deconstruction of unions and aggregations+--  - caching swap+--+-- There is a lot more that could be done (merging the aggregations, etc.)+-- but it is outside the scope of this MVP.++{-| Builds the computation graph by expanding a single node until a transitive+closure is reached.++It performs the naming, node deduplication and cycle detection.++TODO(kps) use the caching information to have a correct fringe+-}+buildComputationGraph :: ComputeNode loc a -> Try ComputeGraph+buildComputationGraph ld = do+  cg <- tryEither $ buildCGraph (untyped ld)+  assignPathsUntyped cg++{-| Performs all the operations that are done on the compute graph:++- fullfilling autocache requests+- checking the cache/uncache pairs+- deconstructions of the unions (in the future)++This could all be done on the server side at this point.+-}+performGraphTransforms :: ComputeGraph -> Try ComputeGraph+performGraphTransforms cg = do+  let g = traceHint "_performGraphTransforms g=" $ computeGraphToGraph cg+  let acg = traceHint "_performGraphTransforms: After autocaching:" $ fillAutoCache cachingType autocacheGen g+  g' <- tryEither acg+  failures <- tryEither $ checkCaching g' cachingType+  case failures of+    [] -> return (graphToComputeGraph g')+    _ -> tryError $ sformat ("Found some caching errors: "%sh) failures++_buildComputation :: SparkSession -> ComputeGraph -> Try Computation+_buildComputation session cg =+  let sid = ssId session+      cid = (ComputationID . pack . show . ssCommandCounter) session+      tiedCg = tieNodes cg+      allNodes = vertexData <$> toList (cdVertices tiedCg)+      terminalNodeNames = nodeName . vertexData <$> toList (cdOutputs tiedCg)+  -- TODO it is missing the first node here, hoping it is the first one.+  in case terminalNodeNames of+    [name] ->+      return $ Computation sid cid allNodes [name] name+    _ -> tryError $ sformat ("Programming error in _build1: cg="%sh) cg++_increaseCompCounter :: SparkStatePure ()+_increaseCompCounter = get >>= \session ->+  let+    curr = ssCommandCounter session+    session2 = session { ssCommandCounter =  curr + 1 }+  in put session2++-- Given an end point, gathers all the nodes reachable from there.+_gatherNodes :: LocalData a -> Try [UntypedNode]+_gatherNodes = tryEither . buildVertexList . untyped++_extractionType :: SQLType a -> SQLType [a]+_extractionType = arrayType . SQLType . unSQLType++-- Like the type, remove the row wrapper in the case of basic elements+-- TODO(kps) figure out what the exact semantics are.+-- It seems collect is behaving differently than the other nodes.+_postprocessBasic :: (HasCallStack) => Cell -> Cell+_postprocessBasic (RowArray rows) =+  RowArray (process <$> rows)  where+    process (RowArray arr) = case V.toList arr of+       [IntElement x] -> IntElement x+       [StringElement x] -> StringElement x+       _ -> RowArray arr+    process x = x+_postprocessBasic x = x+  --failure $ "Could not interpret this cell: " ++ show x++-- Given a result, tries to build the corresponding object out of it+_extract1 :: FinalResult -> SQLType Cell -> Try Cell+_extract1 (Left nf) _ = tryError $ sformat ("got an error "%shown) nf+_extract1 (Right ncs) sqlt = res0 where+  -- Because of the Row semantics, all results are wrappend in a row.+  -- We are using the equivalence between arrays and rows during decoding here.+  wrappingType = _extractionType sqlt+  trow = tryEither $ jsonToCell (unSQLType wrappingType) (ncsData ncs)+  res = trow >>= \l -> case l of+      RowArray arr | V.length arr == 1 -> Right $ _postprocessBasic (V.head arr)+      x -> tryError $ sformat ("ContextInternal:_extract1: Expected on element, got "%shown) x+  res0 = trace ("_extract1: wrappingType = " ++ show wrappingType ++ " ncs = " ++ show ncs ++ " res = " ++ show res) res+++-- Gets the relevant nodes for this computation from this spark session.+-- The computation is assumed to be correct and to contain all the nodes+-- already.+-- TODO: make it a total function+getTargetNodes :: (HasCallStack) => Computation -> [UntypedLocalData]+getTargetNodes comp =+  let+    fun2 :: (HasCallStack) => UntypedNode -> UntypedLocalData+    fun2 n = case asLocalObservable <$> castLocality n of+      Right (Right x) -> x+      err -> failure $ sformat ("_getNodes:fun2: err="%shown%" n="%shown) err n+    finalNodeNames = traceHint "_getTargetNodes: finalNodeNames=" $cTerminalNodes comp+    dct = traceHint "_getTargetNodes: dct=" $ M.fromList $ (nodeName &&& id) <$> cNodes comp+    untyped' = finalNodeNames <&> \n ->+      let err = failure $ sformat ("Could not find "%sh%" in "%sh) n dct+      in M.findWithDefault err n dct+  in fun2 <$> untyped'+++-- Stores the results of the computation in the state (so that we can accelerate the+-- next sessions) and returns the expected final results (as a Cell to be converted)+storeResults :: Computation -> [(LocalData Cell, FinalResult)] -> SparkStatePure (Try Cell)+storeResults comp [] = return e where+  e = tryError $ sformat ("No result returned for computation "%shown) comp+storeResults _ res =+  let+    fun4 :: (LocalData Cell, FinalResult) -> Try Cell+    fun4 (node, fresult) =+      trace ("_storeResults node=" ++ show node ++ "final = " ++ show fresult) $+        _extract1 fresult (nodeType node)+    allResults = sequence $ forM res fun4+    expResult = head allResults -- Just accessing the final result for now+  in+    -- TODO store the results:+    return expResult
+ src/Spark/Core/Internal/ContextStructures.hs view
@@ -0,0 +1,58 @@+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.Internal.ContextStructures(+  SparkSessionConf(..),+  SparkSession(..),+  SparkState,+  SparkStatePure,+  ComputeGraph+) where++import Data.Text(Text)+import Control.Monad.State(StateT, State)+import Control.Monad.Logger(LoggingT)++import Spark.Core.Internal.Client(LocalSessionId)+import Spark.Core.Internal.ComputeDag(ComputeDag)+import Spark.Core.Internal.DatasetStructures(UntypedNode, StructureEdge)++-- | The configuration of a remote spark session in krapsh.+data SparkSessionConf = SparkSessionConf {+ -- | The URL of the end point.+  confEndPoint :: !Text,+  -- | The port used to configure the end point.+  confPort :: !Int,+  -- | (internal) the polling interval+  confPollingIntervalMillis :: !Int,+  -- | (optional) the requested name of the session.+  -- This name must obey a number of rules:+  --  - it must consist in alphanumerical and -,_: [a-zA-Z0-9\-_]+  --  - if it already exists on the server, it will be reconnected to+  --+  -- The default value is "" (a new random context name will be chosen).+  confRequestedSessionName :: !Text+} deriving (Show)++-- | A session in Spark.+-- Encapsualates all the state needed to communicate with Spark+-- and to perfor some simple optimizations on the code.+data SparkSession = SparkSession {+  ssConf :: SparkSessionConf,+  ssId :: LocalSessionId,+  ssCommandCounter :: Integer+} deriving (Show)++-- | Represents the state of a session and accounts for the communication+-- with the server.+type SparkState a = LoggingT (StateT SparkSession IO) a++-- More minimalistic state transforms when doing pure evaluation.+-- (internal type)+type SparkStatePure x = State SparkSession x++{-| internal++A graph of computations. This graph is a direct acyclic graph. Each node is+associated to a global path.+-}+type ComputeGraph = ComputeDag UntypedNode StructureEdge
+ src/Spark/Core/Internal/DAGFunctions.hs view
@@ -0,0 +1,315 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}++{-| A set of utility functions to build and transform DAGs.++Because I could not find a public library for such transforms.++Most krapsh manipulations are converted into graph manipulations.+-}+module Spark.Core.Internal.DAGFunctions(+  DagTry,+  -- Building+  buildGraph,+  buildVertexList,+  buildGraphFromList,+  -- Queries+  graphSinks,+  graphSources,+  -- Transforms+  graphMapVertices,+  graphMapVertices',+  vertexMap,+  graphFlatMapEdges,+  graphMapEdges,+  reverseGraph,+  verticesAndEdges,+) where++import qualified Data.Set as S+import qualified Data.Map.Strict as M+import qualified Data.Vector as V+import Data.List(sortBy)+import Data.Maybe+import Data.Foldable(toList)+import Data.Text(Text)+import Control.Arrow((&&&))+import Control.Monad.Except+import Formatting+import Control.Monad.Identity++import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.Utilities++-- | Separate type of error to make it more general and easier+-- to test.+type DagTry a = Either Text a++{-| Starts from a vertex and expands the vertex to reach all the transitive+closure of the vertex.++Returns a list in lexicographic order of dependencies: the graph corresponding+to this list of elements has one sink (the start element) and potentially+multiple sources. The nodes are ordered so that all the parents are visited+before the node itself.+-}+buildVertexList :: (GraphVertexOperations v, Show v) => v -> DagTry [v]+buildVertexList x = buildVertexListBounded x []++{-| Builds the list of vertices, up to a boundary.+-}+buildVertexListBounded :: (GraphVertexOperations v, Show v) =>+  v -> [v] -> DagTry [v]+buildVertexListBounded x boundary =+  let+    boundaryIds = S.fromList $ vertexToId <$> boundary+    traversals = toList $ _buildList boundaryIds [x] M.empty+    lexico = _lexicographic vertexToId traversals in lexico++-- | Builds a graph by expanding a start vertex.+buildGraph :: forall v e. (GraphOperations v e, Show v, Show e) =>+  v -> DagTry (Graph v e)+buildGraph start = buildVertexList start <&> \vxData ->+  let vertices = [Vertex (vertexToId vx) vx | vx <- vxData]+      -- The edges and vertices are already in the right order, no need+      -- to do further checks+      f :: v -> (VertexId, V.Vector (VertexEdge e v))+      f x =+        let vid = vertexToId x+            g :: (e, v) -> VertexEdge e v+            g (ed, x') =+              let toId = vertexToId x'+                  v' = Vertex toId x'+                  e = Edge vid toId ed+              in VertexEdge v' e+            vedges = g <$> expandVertex x+        in (vid, V.fromList vedges)+      vxs = traceHint "buildGraph: vertices=" $ V.fromList vertices+      edges = traceHint "buildGraph: edges=" $ f <$> vxData+      adj = M.fromList edges+  in Graph adj vxs++{-| Attempts to build a graph from a collection of vertices and edges.++This collection may be invalid (cycles, etc.) and the vertices need not+be in topological order.++All the vertices referred by edges must be present in the list of vertices.+-}+buildGraphFromList :: forall v e. (Show v, Show e) =>+  [Vertex v] -> [Edge e] -> DagTry (Graph v e)+buildGraphFromList vxs eds = do+  -- 1. Group the edges by start point+  -- 2. Find the lexicgraphic order (if possible)+  vxById <- _vertexById vxs+  -- The topological information+  let edTopo = myGroupBy $ (edgeFrom &&& edgeTo) <$> eds+  let vertexById :: VertexId -> DagTry (Vertex v)+      vertexById vid = case M.lookup vid vxById of+        Nothing -> throwError $ sformat ("buildGraphFromList: vertex id found in edge but not in vertices: "%sh) vid+        Just vx -> pure vx+  let f :: Vertex v -> DagTry (Vertex v, [Vertex v])+      f vx =+        let links = M.findWithDefault [] (vertexId vx) edTopo+        in sequence (vertexById <$> links) <&> \l -> (vx, l)+  verticesWithEnds <- sequence $ f <$> vxs+  let indexedVertices = zip [1..] verticesWithEnds <&> \(idx, (vx, l)) -> (idx, vx, l)+  -- The nodes in lexicographic order.+  lexico <- _lexicographic vertexId indexedVertices+  -- Build the edge map:+  -- vertexFromId -> vertexEdge+  let vertexEdge :: Edge e -> DagTry (VertexId, VertexEdge e v)+      vertexEdge e = do+        vxTo <- vertexById (edgeTo e)+        -- Used to confirm that the start vertex is here+        _ <- vertexById (edgeFrom e)+        return (edgeFrom e, VertexEdge vxTo e)+  vEdges <- sequence $ vertexEdge <$> eds+  let edgeMap = M.map V.fromList (myGroupBy vEdges)+  return $ Graph edgeMap (V.fromList lexico)++_vertexById :: (Show v) => [Vertex v] -> DagTry (M.Map VertexId (Vertex v))+_vertexById vxs =+  -- This is probably not the most pretty, but it works.+  let vxById = myGroupBy $ (vertexId &&& id) <$> vxs+      f (vid, [vx]) = pure (vid, vx)+      f (vid, l) = throwError $ sformat ("_VertexById: Multiple vertices with the same id: "%sh%" in "%sh) vid l+  in M.fromList <$> sequence (f <$> M.toList vxById)++-- This implementation is not very efficient and is probably a performance+-- bottleneck.+-- Int is the traversal order. It is just used to break the ties.+-- VertexId is the node id of the vertex.+_lexicographic :: (v -> VertexId) -> [(Int, v, [v])] -> DagTry [v]+_lexicographic _ [] = return []+_lexicographic f m =+  -- We use the traversal ordering to separate the ties.+  -- The first nodes traversed get priority.+  let fcmp (idx, _, []) (idx', _, []) = compare idx idx'+      fcmp (_, _, []) (_, _, _) = LT+      fcmp (_, _, _) (_, _, []) = GT+      fcmp (_, _, _) (_, _, _) = EQ -- This one does not matter+  in case sortBy fcmp m of+    [] -> throwError "_lexicographic: there is a cycle"+    ((_, v, _) : t) ->+      let currentId = f v+          removeCurrentId l = [v' | v' <- l, f v' /= currentId]+          m' = t <&> \(idx, v', l) -> (idx, v', removeCurrentId l)+          tl = _lexicographic f m'+      in (v :) <$> tl+++_buildList :: (Show v, GraphVertexOperations v) =>+  S.Set VertexId -> -- boundary ids, they will not be traversed+  [v] -> -- fringe ids+  M.Map VertexId (Int, v, [v]) -> -- all seen ids so far (the intermediate result)+  M.Map VertexId (Int, v, [v])+_buildList boundary fringe =+  _buildListGeneral boundary fringe expandVertexAsVertices++-- (internal)+-- Gathers the list of all the nodes connected through this graph+--+-- The expansion function in that case can be controlled.+--+-- The expansion is done in a DFS manner (the order of the node is unique).+_buildListGeneral :: (Show v, GraphVertexOperations v) =>+  S.Set VertexId -> -- boundary ids, they will not be traversed+  [v] -> -- fringe ids: the nodes that have been touched but not expanded.+  (v -> [v]) -> -- The expansion function. They will be the next nodes to expand.+  -- all seen ids so far (the intermediate result)+  -- along with the index of the node during the traversal, and the+  -- node itself.+  M.Map VertexId (Int, v, [v]) ->+  M.Map VertexId (Int, v, [v])+_buildListGeneral _ [] _ allSeen = allSeen+_buildListGeneral boundaryIds (x : t) expand allSeen =+  let vid = vertexToId x in+  if M.member vid allSeen || S.member vid boundaryIds then+    _buildListGeneral boundaryIds t expand allSeen+  else+    let nextVertices = expand x+        currIdx = M.size allSeen+        allSeen2 = M.insert vid (currIdx, x, nextVertices) allSeen+        filterFun y = not $ M.member (vertexToId y) allSeen2+        nextVertices2 = filter filterFun nextVertices+    in _buildListGeneral boundaryIds (nextVertices2 ++ t) expand allSeen2++{-| The sources of a DAG (nodes with no parent).+-}+graphSources :: Graph v e -> [Vertex v]+graphSources g =+  let hasParent = do+        vedges <- toList (gEdges g)+        edge <- toList vedges+        return . vertexId . veEndVertex $ edge+      hasPSet = S.fromList hasParent+      -- false iff the vertex has an incoming edge+      filt vx = not (S.member (vertexId vx) hasPSet)+  in filter filt (toList (gVertices g))++{-| The sinks of a graph (nodes with no descendant).+-}+graphSinks :: Graph v e -> [Vertex v]+graphSinks g =+  let f vx = V.null (M.findWithDefault V.empty (vertexId vx) (gEdges g))+  in filter f (toList (gVertices g))++-- | Flips the edges of this graph (it is also a DAG)+reverseGraph :: forall v e. (HasCallStack, Show v, Show e) => Graph v e -> Graph v e+reverseGraph g =+  let+    vxMap = M.fromList ((vertexId &&& id) <$> toList (gVertices g))+    flipVEdge :: (VertexId, V.Vector (VertexEdge e v)) -> [(VertexId, VertexEdge e v)]+    flipVEdge (fromNid, vec) = case M.lookup fromNid vxMap of+      Nothing -> [] -- Should be a programming error+      Just endVx ->+        toList vec <&> \ve ->+          let ed = veEdge ve+              oldEndVx = veEndVertex ve+              oldEndVid = vertexId oldEndVx+              ed' = Edge {+                edgeFrom = oldEndVid,+                edgeTo = fromNid,+                edgeData = edgeData ed }+          in (oldEndVid, VertexEdge { veEdge = ed', veEndVertex = endVx })+    edges = myGroupBy $ concat $ flipVEdge <$> M.toList (gEdges g)+  in Graph (V.fromList <$> edges) (V.reverse (gVertices g))++-- | A generic transform over the graph that may account for potential failures+-- in the process.+graphMapVertices :: forall m v e v2. (HasCallStack, Show v2, Show v, Show e, Monad m) =>+  Graph v e -> -- The start graph+  (v -> [(v2,e)] -> m v2) -> -- The transform+  m (Graph v2 e)+graphMapVertices g f =+  let+    fun :: M.Map VertexId v2 -> [Vertex v] -> m [Vertex v2]+    fun _ [] = return []+    fun done (vx : t) =+      let+        vid = vertexId vx+        parents = V.toList $ fromMaybe V.empty $ M.lookup vid (gEdges g)+        parentEdges = veEdge <$> parents+        getPairs :: Edge e -> (v2, e)+        getPairs ed =+          let vidTo = edgeTo ed+              msg = sformat ("graphMapVertices: Could not locate "%shown%" in "%shown)vidTo done+              -- The edges are flowing from child -> parent so+              -- to == parent+              vert = fromMaybe (failure msg) (M.lookup vidTo done)+            in (vert, edgeData ed)+        parents2 = [getPairs ed | ed <- parentEdges]+        -- parents2 = [fromJust $ M.lookup vidFrom done | vidFrom <- parentVids]+        merge0 :: v2 -> m [Vertex v2]+        merge0 vx2Data =+          let done2 = M.insert vid vx2Data done+              vx2 = vx { vertexData = vx2Data }+              rest = fun done2 t in+            (vx2 : ) <$> rest+      in+        f (vertexData vx) parents2 >>= merge0+  in do+    verts2 <- fun M.empty (toList (gVertices g))+    let+      idxs2 = M.fromList [(vertexId vx2, vx2) | vx2 <- verts2]+      trans :: Vertex v -> Vertex v2+      trans vx = fromJust $ M.lookup (vertexId vx) idxs2+      conv :: VertexEdge e v -> VertexEdge e v2+      conv (VertexEdge vx1 e1) = VertexEdge (trans vx1) e1+      adj2 = M.map (conv <$>) (gEdges g)+    return Graph { gEdges = adj2, gVertices = V.fromList verts2 }++-- | (internal) Maps the edges+graphMapEdges :: Graph v e -> (e -> e') -> Graph v e'+graphMapEdges g f = graphFlatMapEdges g ((:[]) . f)++-- | (internal) Maps and the edges, and may create more or less.+graphFlatMapEdges :: Graph v e -> (e -> [e']) -> Graph v e'+graphFlatMapEdges g f = g { gEdges = edges } where+  fun (VertexEdge vx ed) =+    f (edgeData ed) <&> \ed' -> VertexEdge vx (ed { edgeData = ed' })+  edges = (V.fromList . concatMap fun) <$> gEdges g++-- | (internal) Maps the vertices.+graphMapVertices' :: (Show v, Show e, Show v') => (v -> v') -> Graph v e -> Graph v' e+graphMapVertices' f g =+  runIdentity (graphMapVertices g f') where+    f' v _ = return $ f v++-- | The map of vertices, by vertex id.+vertexMap :: Graph v e -> M.Map VertexId v+vertexMap g =+  M.fromList . toList $ gVertices g <&> (vertexId &&& vertexData)++-- (internal)+-- The vertices in lexicographic order, and the originating edges for these+-- vertices.+verticesAndEdges :: Graph v e -> [([(v, e)],v)]+verticesAndEdges g =+  toList (gVertices g) <&> \vx ->+    let n = vertexData vx+        l = V.toList $ M.findWithDefault V.empty (vertexId vx) (gEdges g)+        lres = [(vertexData vx', edgeData e') | (VertexEdge vx' e') <- l]+    in (lres, n)
+ src/Spark/Core/Internal/DAGStructures.hs view
@@ -0,0 +1,107 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}++{-| Data structures to represent Directed Acyclic Graphs (DAGs).++-}+module Spark.Core.Internal.DAGStructures where++import qualified Data.Map.Strict as M+import qualified Data.Text as T+import Data.ByteString(ByteString)+import Data.Vector(Vector)+import Data.Foldable(toList)+import Data.Hashable(Hashable)+import GHC.Generics(Generic)+import Formatting++import Spark.Core.Internal.Utilities++-- | The unique ID of a vertex.+newtype VertexId = VertexId { unVertexId :: ByteString } deriving (Eq, Ord, Generic)+++-- | An edge in a graph, parametrized by some payload.+data Edge e = Edge {+  edgeFrom :: !VertexId,+  edgeTo :: !VertexId,+  edgeData :: !e+}++-- | A vertex in a graph, parametrized by some payload.+data Vertex v = Vertex {+  vertexId :: !VertexId,+  vertexData :: !v+}++{-| An edge, along with its end node.+-}+data VertexEdge e v = VertexEdge {+    veEndVertex :: !(Vertex v),+    veEdge :: !(Edge e) }++{-| The adjacency map of a graph.++The node Id corresponds to the start node, the pairs are the end node and+and the edge to reach to the node. There may be multiple edges leading to the+same node.+-}+type AdjacencyMap v e = M.Map VertexId (Vector (VertexEdge e v))++-- | The representation of a graph.+--+-- In all the project, it is considered as a DAG.+data Graph v e = Graph {+  gEdges :: !(AdjacencyMap v e),+  gVertices :: !(Vector (Vertex v))+}++-- | Graph operations on types that are supposed to+-- represent vertices.+class GraphVertexOperations v where+  vertexToId :: v -> VertexId+  expandVertexAsVertices :: v -> [v]++-- | Graph operations on types that are supposed to represent+-- edges.+class (GraphVertexOperations v) => GraphOperations v e where+  expandVertex :: v -> [(e,v)]++instance Functor Vertex where+  fmap f vx = vx { vertexData = f (vertexData vx) }++instance Functor Edge where+  fmap f ed = ed { edgeData = f (edgeData ed) }++instance (Show v) => Show (Vertex v) where+  show vx = "Vertex(vId=" ++ show (vertexId vx) ++ " v=" ++ show (vertexData vx) ++ ")"++instance (Show e) => Show (Edge e) where+  show ed = "Edge(from=" ++ show (edgeFrom ed) ++ " to=" ++ show (edgeTo ed) ++ " e=" ++ show (edgeData ed) ++ ")"++instance (Show v, Show e) => Show (VertexEdge e v) where+  show (VertexEdge v e) = "(" ++ show v ++ ", " ++ show e ++ ")"++instance (Show v, Show e) => Show (Graph v e) where+  show g =+    let vxs = toList $ gVertices g <&> \(Vertex vid x) ->+          sformat (sh%":"%sh) vid x+        vedges = foldMap toList (M.elems (gEdges g))+        edges = (veEdge <$> vedges) <&> \(Edge efrom eto x) ->+          sformat (sh%"->"%sh%"->"%sh) efrom x eto+        -- eds = (M.elems (gEdges g)) `foldMap` \v ->+        --   (toList v) <&>+        vxs' = T.intercalate "," vxs+        eds' = T.intercalate " " edges+        str = T.concat ["Graph{", vxs', ", ", eds', "}"]+    in T.unpack str++instance Hashable VertexId++instance Show VertexId where+  show (VertexId bs) = let s = show bs in+    if length s > 9 then+      (drop 1 . take 6) s ++ ".."+    else+      s
+ src/Spark/Core/Internal/DatasetFunctions.hs view
@@ -0,0 +1,509 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}++module Spark.Core.Internal.DatasetFunctions(+  parents,+  untyped,+  logicalParents,+  depends,+  dataframe,+  asDF,+  asDS,+  asLocalObservable,+  -- Standard functions+  identity,+  autocache,+  cache,+  uncache,+  union,+  -- Developer+  castLocality,+  emptyDataset,+  emptyLocalData,+  emptyNodeStandard,+  nodeId,+  nodeLogicalDependencies,+  nodeLogicalParents,+  nodeLocality,+  nodeName,+  nodePath,+  nodeOp,+  nodeParents,+  nodeType,+  untypedDataset,+  updateNode,+  -- Developer conversions+  fun1ToOpTyped,+  fun2ToOpTyped,+  nodeOpToFun1,+  nodeOpToFun1Typed,+  nodeOpToFun1Untyped,+  nodeOpToFun2,+  nodeOpToFun2Typed,+  -- Internal+  opnameCache,+  opnameUnpersist,+  opnameAutocache,++) where++import qualified Crypto.Hash.SHA256 as SHA+import qualified Data.Aeson as A+import qualified Data.Text as T+import qualified Data.Text.Format as TF+import qualified Data.Vector as V+import Data.Aeson((.=), toJSON)+import Data.Text.Encoding(decodeUtf8)+import Data.ByteString.Base16(encode)+import Data.Maybe(fromMaybe, listToMaybe)+import Data.Text.Lazy(toStrict)+import Data.String(IsString(fromString))+import Formatting++import Spark.Core.StructuresInternal+import Spark.Core.Try+import Spark.Core.Row+import Spark.Core.Internal.TypesStructures+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.OpStructures+import Spark.Core.Internal.OpFunctions+import Spark.Core.Internal.Utilities+import Spark.Core.Internal.RowUtils+import Spark.Core.Internal.TypesGenerics++-- | (developer) The operation performed by this node.+nodeOp :: ComputeNode loc a -> NodeOp+nodeOp = _cnOp++-- | The nodes this node depends on directly.+nodeParents :: ComputeNode loc a -> [UntypedNode]+nodeParents = V.toList . _cnParents++-- | (developer) Returns the logical parenst of a node.+nodeLogicalParents :: ComputeNode loc a -> Maybe [UntypedNode]+nodeLogicalParents = (V.toList <$>) . _cnLogicalParents++-- | Returns the logical dependencies of a node.+nodeLogicalDependencies :: ComputeNode loc a -> [UntypedNode]+nodeLogicalDependencies = V.toList . _cnLogicalDeps++-- | The name of a node.+-- TODO: should be a NodePath+nodeName :: ComputeNode loc a -> NodeName+nodeName node = fromMaybe (_defaultNodeName node) (_cnName node)++{-| The path of a node, as resolved.++This path includes information about the logical parents (after resolution).+-}+nodePath :: ComputeNode loc a -> NodePath+nodePath node =+  if V.null . unNodePath . _cnPath $ node+    then NodePath . V.singleton . nodeName $ node+    else _cnPath node++-- | The type of the node+-- TODO have nodeType' for dynamic types as well+nodeType :: ComputeNode loc a -> SQLType a+nodeType = SQLType . _cnType++{-| The identity function.++Returns a compute node with the same datatype and the same content as the+previous node. If the operation of the input has a side effect, this side+side effect is *not* reevaluated.++This operation is typically used when establishing an ordering between some+operations such as caching or side effects, along with `logicalDependencies`.+-}+identity :: ComputeNode loc a -> ComputeNode loc a+identity n = n2 `parents` [untyped n]+  where n2 = emptyNodeStandard (nodeLocality n) (nodeType n) name+        name = if _cnLocality n == Local+                then "org.spark.LocalIdentity"+                else "org.spark.Identity"++{-| Caches the dataset.++This function instructs Spark to cache a dataset with the default persistence+level in Spark (MEMORY_AND_DISK).++Note that the dataset will have to be evaluated first for the caching to take+effect, so it is usual to call `count` or other aggregrators to force+the caching to occur.+-}+cache :: Dataset a -> Dataset a+cache  n = n2 `parents` [untyped n]+  where n2 = emptyNodeStandard (nodeLocality n) (nodeType n) opnameCache++-- (internal)+opnameCache :: T.Text+opnameCache = "org.spark.Cache"++{-| Uncaches the dataset.++This function instructs Spark to unmark the dataset as cached. The disk and the+memory used by Spark in the future.++Unlike Spark, Krapsh is stricter with the uncaching operation:+ - the argument of cache must be a cached dataset+ - once a dataset is uncached, its cached version cannot be used again (i.e. it+   must be recomputed).++Krapsh performs escape analysis and will refuse to run programs with caching+issues.+-}+uncache :: ComputeNode loc a -> ComputeNode loc a+uncache  n = n2 `parents` [untyped n]+  where n2 = emptyNodeStandard (nodeLocality n) (nodeType n) opnameUnpersist++-- (internal)+opnameUnpersist :: T.Text+opnameUnpersist = "org.spark.Unpersist"++{-| Automatically caches the dataset on a need basis, and performs deallocation+when the dataset is not required.++This function marks a dataset as eligible for the default caching level in+Spark. The current implementation performs caching only if it can be established+that the dataset is going to be involved in more than one shuffling or+aggregation operation.++If the dataset has no observable child, no uncaching operation is added: the+autocache operation is equivalent to unconditional caching.+-}+autocache :: Dataset a -> Dataset a+autocache n = n2 `parents` [untyped n]+  where n2 = emptyNodeStandard (nodeLocality n) (nodeType n) opnameAutocache++opnameAutocache :: T.Text+opnameAutocache = "org.spark.Autocache"++{-| Returns the union of two datasets.++In the context of streaming and differentiation, this union is biased towards+the left: the left argument expresses the stream and the right element expresses+the increment.+-}+union :: Dataset a -> Dataset a -> Dataset a+union n1 n2 = n `parents` [untyped n1, untyped n2]+  where n = emptyNodeStandard (nodeLocality n1) (nodeType n1) _opnameUnion++_opnameUnion :: T.Text+_opnameUnion = "org.spark.Union"++-- | Converts to a dataframe and drops the type info.+-- This always works.+asDF :: ComputeNode LocDistributed a -> DataFrame+asDF = pure . _unsafeCastNode++-- | Attempts to convert a dataframe into a (typed) dataset.+--+-- This will fail if the dataframe itself is a failure, of if the casting+-- operation is not correct.+-- This operation assumes that both field names and types are correct.+asDS :: forall a. (SQLTypeable a) => DataFrame -> Try (Dataset a)+asDS df = do+  n <- df+  let dt = unSQLType (buildType :: SQLType a)+  let dt' = unSQLType (nodeType n)+  if dt == dt'+    then pure (_unsafeCastNode n)+    else tryError $ sformat ("Casting error: dataframe has type "%sh%" incompatible with type "%sh) dt' dt+++-- | Converts a local node to a local frame.+-- This always works.+asLocalObservable :: ComputeNode LocLocal a -> LocalFrame+asLocalObservable = pure . _unsafeCastNode++-- | Converts any node to an untyped node+untyped :: ComputeNode loc a -> UntypedNode+untyped = _unsafeCastNode++untypedDataset :: ComputeNode LocDistributed a -> UntypedDataset+untypedDataset = _unsafeCastNode++{-| Adds parents to the node.+It is assumed the parents are the unique set of nodes required+by the operation defined in this node.+If you want to set parents for the sake of organizing computation+use logicalParents.+If you want to add some timing dependencies between nodes,+use depends.+-}+parents :: ComputeNode loc a -> [UntypedNode] -> ComputeNode loc a+parents node l = updateNode node $ \n ->+  n { _cnParents = V.fromList l V.++ _cnParents n }++{-| Establishes a naming convention on this node: the path of this node will be+determined as if the parents of this node were the list provided (and without+any effect from the direct parents of this node).++For this to work, the logical parents should split the nodes between internal+nodes, logical parents, and the rest. In other words, for any ancestor of this node,+and for any valid path to reach this ancestor, this path should include at least one+node from the logical dependencies.++This set can be a super set of the actual logical parents.++The check is lazy (done during the analysis phase). An error (if any) will+only be reported during analysis.+-}+logicalParents :: ComputeNode loc a -> [UntypedNode] -> ComputeNode loc a+logicalParents node l = updateNode node $ \n ->+  n { _cnLogicalParents = pure . V.fromList $ l }++{-| Sets the logical dependencies on this node.++All the nodes given will be guaranteed to be executed before the current node.++If there are any failures, this node will also be treated as a failure (even+if the parents are all successes).+-}+depends :: ComputeNode loc a -> [UntypedNode] -> ComputeNode loc a+depends node l = updateNode node $ \n ->+  n { _cnLogicalDeps = V.fromList l }+++-- (internal)+-- Tries to update the locality of a node. This is a checked cast.+-- TODO: remove, it is only used to cast to local frame+castLocality :: forall a loc loc'. (+  CheckedLocalityCast loc, CheckedLocalityCast loc') =>+    ComputeNode loc a -> Try (ComputeNode loc' a)+castLocality node =+  let+    loc2 = _cnLocality node+    locs = unTypedLocality <$> (_validLocalityValues :: [TypedLocality loc'])+  in if locs == [loc2] then+    pure $ node { _cnLocality = loc2 }+  else+    tryError $ sformat ("Wrong locality :"%shown%", expected: "%shown) loc2 locs++-- (internal)+-- The id of a node. If it is not set in the node, it will be+-- computed from scratch.+-- This is a potentially long operation.+nodeId :: ComputeNode loc a -> NodeId+nodeId = _cnNodeId++-- (internal)+-- This operation should always be used to make sure that the+-- various caches inside the compute node are maintained.+updateNode :: ComputeNode loc a -> (ComputeNode loc a -> ComputeNode loc' b) -> ComputeNode loc' b+updateNode ds f = ds2 { _cnNodeId = id2 } where+  ds2 = f ds+  id2 = _nodeId ds2+++-- (internal)+-- The locality of the node+nodeLocality :: ComputeNode loc a -> TypedLocality loc+nodeLocality = TypedLocality . _cnLocality++-- (internal)+emptyDataset :: NodeOp -> SQLType a -> Dataset a+emptyDataset = _emptyNode++-- (internal)+emptyLocalData :: NodeOp -> SQLType a -> LocalData a+emptyLocalData = _emptyNode++{-| Creates a dataframe from a list of cells and a datatype.++Wil fail if the content of the cells is not compatible with the+data type.+-}+dataframe :: DataType -> [Cell] -> DataFrame+dataframe dt cells' = do+  validCells <- tryEither $ sequence (checkCell dt <$> cells')+  let jData = V.fromList (toJSON <$> validCells)+  let op = NodeDistributedLit dt jData+  return $ _emptyNode op (SQLType dt)+++-- *********** function / object conversions *******++-- | (internal)+placeholderTyped :: forall a loc. (IsLocality loc) =>+  SQLType a -> ComputeNode loc a+placeholderTyped tp =+  let+    so = makeOperator "org.spark.Placeholder" tp+    (TypedLocality l) = _getTypedLocality :: TypedLocality loc+    op = case l of+      Local -> NodeLocalOp so+      Distributed -> NodeDistributedOp so+  in  _emptyNode op tp++-- | (internal) conversion+fun1ToOpTyped :: forall a loc a' loc'. (IsLocality loc) =>+  SQLType a -> (ComputeNode loc a -> ComputeNode loc' a') -> NodeOp+fun1ToOpTyped sqlt f = nodeOp $ f (placeholderTyped sqlt)++-- | (internal) conversion+fun2ToOpTyped :: forall a1 a2 a loc1 loc2 loc. (IsLocality loc1, IsLocality loc2) =>+  SQLType a1 -> SQLType a2 -> (ComputeNode loc1 a1 -> ComputeNode loc2 a2 -> ComputeNode loc a) -> NodeOp+fun2ToOpTyped sqlt1 sqlt2 f = nodeOp $ f (placeholderTyped sqlt1) (placeholderTyped sqlt2)++-- | (internal) conversion+nodeOpToFun1 :: forall a1 a2 loc1 loc2. (IsLocality loc1, SQLTypeable a2, IsLocality loc2) =>+  NodeOp -> ComputeNode loc1 a1 -> ComputeNode loc2 a2+nodeOpToFun1 = nodeOpToFun1Typed (buildType :: SQLType a2)++-- | (internal) conversion+nodeOpToFun1Typed :: forall a1 a2 loc1 loc2. (HasCallStack, IsLocality loc1, IsLocality loc2) =>+  SQLType a2 -> NodeOp -> ComputeNode loc1 a1 -> ComputeNode loc2 a2+nodeOpToFun1Typed sqlt no node =+  let n2 = _emptyNode no sqlt :: ComputeNode loc2 a2+  in n2 `parents` [untyped node]++-- | (internal) conversion+nodeOpToFun1Untyped :: forall loc1 loc2. (HasCallStack, IsLocality loc1, IsLocality loc2) =>+  DataType -> NodeOp -> ComputeNode loc1 Cell -> ComputeNode loc2 Cell+nodeOpToFun1Untyped dt no node =+  let n2 = _emptyNode no (SQLType dt) :: ComputeNode loc2 Cell+  in n2 `parents` [untyped node]++-- | (internal) conversion+nodeOpToFun2 :: forall a a1 a2 loc loc1 loc2. (SQLTypeable a, IsLocality loc, IsLocality loc1, IsLocality loc2) =>+  NodeOp -> ComputeNode loc1 a1 -> ComputeNode loc2 a2 -> ComputeNode loc a+nodeOpToFun2 = nodeOpToFun2Typed (buildType :: SQLType a)++-- | (internal) conversion+nodeOpToFun2Typed :: forall a a1 a2 loc loc1 loc2. (IsLocality loc, IsLocality loc1, IsLocality loc2) =>+  SQLType a -> NodeOp -> ComputeNode loc1 a1 -> ComputeNode loc2 a2 -> ComputeNode loc a+nodeOpToFun2Typed sqlt no node1 node2 =+  let n2 = _emptyNode no sqlt :: ComputeNode loc a+  in n2 `parents` [untyped node1, untyped node2]++-- ******* INSTANCES *********++-- Put here because it depends on some private functions.+instance forall loc a. Show (ComputeNode loc a) where+  show ld = let+    txt = fromString "{}@{}{}{{}}" :: TF.Format+    loc :: T.Text+    loc = case nodeLocality ld of+      TypedLocality Local -> "!"+      TypedLocality Distributed -> ""+    nn = unNodeName . nodeName $ ld+    no = simpleShowOp . nodeOp $ ld+    fields = T.pack . show . nodeType $ ld in+      T.unpack $ toStrict $ TF.format txt (nn, no, loc, fields)++instance forall loc a. A.ToJSON (ComputeNode loc a) where+  toJSON node = A.object [+    "locality" .= nodeLocality node,+    "name" .= nodeName node,+    "op" .= (simpleShowOp . nodeOp $ node),+    "extra" .= (extraNodeOpData . nodeOp $ node),+    "parents" .= (nodeName <$> nodeParents node),+    "logicalDependencies" .= (nodeName <$> nodeLogicalDependencies node),+    "_type" .= (unSQLType . nodeType) node]++instance forall loc. A.ToJSON (TypedLocality loc) where+  toJSON (TypedLocality Local) = A.String "local"+  toJSON (TypedLocality Distributed) = A.String "distributed"++-- Performs an unsafe type recast.+-- This is useful for internal code that knows whether+-- this operation is legal or not through some other means.+-- This may still throw an error if the cast is illegal.+_unsafeCastNode :: CheckedLocalityCast loc2 => ComputeNode loc1 a -> ComputeNode loc2 b+_unsafeCastNode x = x {+    _cnType = _cnType x,+    _cnLocality = _cnLocality x+  }++_unsafeCastNodeTyped :: TypedLocality loc2 -> ComputeNode loc1 a -> ComputeNode loc2 b+_unsafeCastNodeTyped l x = x {+    _cnType = _cnType x,+    _cnLocality = unTypedLocality l+  }++--+_unsafeCastLoc :: CheckedLocalityCast loc' =>+  TypedLocality loc -> TypedLocality loc'+_unsafeCastLoc (TypedLocality Local) =+  checkLocalityValidity (TypedLocality Local)+_unsafeCastLoc (TypedLocality Distributed) =+  checkLocalityValidity (TypedLocality Distributed)+++-- This should be a programming error+checkLocalityValidity :: forall loc. (HasCallStack, CheckedLocalityCast loc) =>+  TypedLocality loc -> TypedLocality loc+checkLocalityValidity x =+  if x `notElem` _validLocalityValues+    then+      let msg = sformat ("CheckedLocalityCast: element "%shown%" not in the list of accepted values: "%shown)+                  x (_validLocalityValues :: [TypedLocality loc])+      in failure msg x+    else x+++-- Computes the ID of a node.+-- Since this is a complex operation, it should be cached by each node.+_nodeId :: ComputeNode loc a -> NodeId+_nodeId node =+  let c1 = SHA.init+      f2 = unNodeId . nodeId+      c2 = hashUpdateNodeOp c1 (nodeOp node)+      c3 = SHA.updates c2 $ f2 <$> nodeParents node+      c4 = SHA.updates c3 $ f2 <$> nodeLogicalDependencies node+      -- c6 = SHA.update c4 $ (BS.concat . LBS.toChunks) b+  in+    -- Using base16 encoding to make sure it is readable.+    -- Not sure if it is a good idea in general.+    (NodeId . encode . SHA.finalize) c4++_defaultNodeName :: ComputeNode loc a -> NodeName+_defaultNodeName node =+  let opName = (simpleShowOp . nodeOp) node+      namePieces = T.splitOn (T.pack ".") opName+      lastOpt = (listToMaybe . reverse) namePieces+      l = fromMaybe (T.pack "???") lastOpt+      idbs = nodeId node+      idt = (T.take 6 . decodeUtf8 . unNodeId) idbs+      n = T.concat [T.toLower l, T.pack "_", idt]+  in NodeName n++-- Create a new empty node. Also performs a locality check to+-- make sure the info being provided is correct.+_emptyNode :: forall loc a. (IsLocality loc) =>+  NodeOp -> SQLType a -> ComputeNode loc a+_emptyNode op sqlt = _emptyNodeTyped (_getTypedLocality :: TypedLocality loc) sqlt op++_emptyNodeTyped :: forall loc a.+  TypedLocality loc -> SQLType a -> NodeOp -> ComputeNode loc a+_emptyNodeTyped tloc (SQLType dt) op = updateNode (_unsafeCastNodeTyped tloc ds) id where+  ds :: ComputeNode loc a+  ds = ComputeNode {+    _cnName = Nothing,+    _cnOp = op,+    _cnType = dt,+    _cnParents = V.empty,+    _cnLogicalParents = Nothing,+    _cnLogicalDeps = V.empty,+    _cnLocality = unTypedLocality tloc,+    _cnNodeId = error "_emptyNode: _cnNodeId",+    _cnPath = NodePath V.empty+  }++emptyNodeStandard :: forall loc a.+  TypedLocality loc -> SQLType a -> T.Text -> ComputeNode loc a+emptyNodeStandard tloc sqlt name = _emptyNodeTyped tloc sqlt op where+  so = StandardOperator {+         soName = name,+         soOutputType = unSQLType sqlt,+         soExtra = A.Null+       }+  op = if unTypedLocality tloc == Local+          then NodeLocalOp so+          else NodeDistributedOp so
+ src/Spark/Core/Internal/DatasetStructures.hs view
@@ -0,0 +1,198 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE FlexibleInstances #-}++module Spark.Core.Internal.DatasetStructures where++import qualified Data.Text as T+import Data.Vector(Vector)++import Spark.Core.StructuresInternal+import Spark.Core.Try+import Spark.Core.Row+import Spark.Core.Internal.Utilities+import Spark.Core.Internal.OpStructures+import Spark.Core.Internal.TypesStructures++{-| (internal) The main data structure that represents a data node in the+computation graph.++This data structure forms the backbone of computation graphs expressed+with spark operations.++loc is a typed locality tag.+a is the type of the data, as seen by the Haskell compiler. If erased, it+will be a Cell type.+-}+-- TODO: separate the topology info from the node info. It will help when+-- building the graphs.+data ComputeNode loc a = ComputeNode {+  -- | The id of the node.+  --+  -- Non strict because it may be expensive.+  _cnNodeId :: NodeId,+  -- The following fields are used to build a unique ID to+  -- a compute node:++  -- | The operation associated to this node.+  _cnOp :: !NodeOp,+  -- | The type of the node+  _cnType :: !DataType,+  -- | The direct parents of the node. The order of the parents is important+  -- for the semantics of the operation.+  _cnParents :: !(Vector UntypedNode),+  -- | A set of extra dependencies that can be added to force an order between+  -- the nodes.+  --+  -- The order is not important, they are sorted by ID.+  --+  -- TODO(kps) add this one to the id+  _cnLogicalDeps :: !(Vector UntypedNode),+  -- | The locality of this node.+  --+  -- TODO(kps) add this one to the id+  _cnLocality :: !Locality,+  -- Attributes that are not included in the id+  -- These attributes are mostly for the user to relate to the nodes.+  -- They are not necessary for the computation.+  --+  -- | The name+  _cnName :: !(Maybe NodeName),+  -- | A set of nodes considered as the logical input for this node.+  -- This has no influence on the calculation of the id and is used+  -- for organization purposes only.+  _cnLogicalParents :: !(Maybe (Vector UntypedNode)),+  -- | The path of this oned in a computation flow.+  --+  -- This path includes the node name.+  -- Not strict because it may be expensive to compute.+  -- By default it only contains the name of the node (i.e. the node is+  -- attached to the root)+  _cnPath :: NodePath+} deriving (Eq)++-- (internal) Phantom type tags for the locality+data TypedLocality loc = TypedLocality { unTypedLocality :: !Locality } deriving (Eq, Show)+data LocLocal+data LocDistributed+data LocUnknown++-- (developer) The type for which we drop all the information expressed in+-- types.+--+-- This is useful to express parent dependencies (pending a more type-safe+-- interface)+type UntypedNode = ComputeNode LocUnknown Cell++-- (internal) A dataset for which we have dropped type information.+-- Used internally by columns.+type UntypedDataset = Dataset Cell++type UntypedLocalData = LocalData Cell++{-| A typed collection of distributed data.++Most operations on datasets are type-checked by the Haskell+compiler: the type tag associated to this dataset is guaranteed+to be convertible to a proper Haskell type. In particular, building+a Dataset of dynamic cells is guaranteed to never happen.++If you want to do untyped operations and gain+some flexibility, consider using UDataFrames instead.++Computations with Datasets and observables are generally checked for+correctness using the type system of Haskell.+-}+type Dataset a = ComputeNode LocDistributed a++{-|+A unit of data that can be accessed by the user.++This is a typed unit of data. The type is guaranteed to be a proper+type accessible by the Haskell compiler (instead of simply a Cell+type, which represents types only accessible at runtime).++TODO(kps) rename to Observable+-}+type LocalData a = ComputeNode LocLocal a+++{-|+The dataframe type. Any dataset can be converted to a dataframe.++For the Spark users: this is different than the definition of the+dataframe in Spark, which is a dataset of rows. Because the support+for single columns is more akward in the case of rows, it is more+natural to generalize datasets to contain cells.+When communicating with Spark, though, single cells are wrapped+into rows with single field, as Spark does.+-}+type DataFrame = Try (Dataset Cell)++{-| Observable, whose type can only be infered at runtime and+that can fail to be computed at runtime.++Any observable can be converted to an untyped+observable.++Untyped observables are more flexible and can be combined in+arbitrary manner, but they will fail during the validation of+the Spark computation graph.++TODO(kps) rename to DynObservable+-}+type LocalFrame = Try (LocalData Cell)++{-| The different paths of edges in the compute DAG of nodes, at the+start of computations.++ - scope edges specify the scope of a node for naming. They are not included in+   the id.++-}+data NodeEdge = ScopeEdge | DataStructureEdge StructureEdge deriving (Show, Eq)++{-| The edges in a compute DAG, after name resolution (which is where most of+the checks and computations are being done)++- parent edges are the direct parents of a node, the only ones required for+  defining computations. They are included in the id.+- logical edges define logical dependencies between nodes to force a specific+  ordering of the nodes. They are included in the id.+-}+data StructureEdge = ParentEdge | LogicalEdge deriving (Show, Eq)+++class CheckedLocalityCast loc where+  _validLocalityValues :: [TypedLocality loc]++-- Class to retrieve the locality associated to a type.+-- Is it better to use type classes?+class (CheckedLocalityCast loc) => IsLocality loc where+  _getTypedLocality :: TypedLocality loc++instance CheckedLocalityCast LocLocal where+  _validLocalityValues = [TypedLocality Local]++instance CheckedLocalityCast LocDistributed where+  _validLocalityValues = [TypedLocality Distributed]++-- LocLocal is a locality associated to Local+instance IsLocality LocLocal where+  _getTypedLocality = TypedLocality Local++-- LocDistributed is a locality associated to Distributed+instance IsLocality LocDistributed where+  _getTypedLocality = TypedLocality Distributed++instance CheckedLocalityCast LocUnknown where+  _validLocalityValues = [TypedLocality Distributed, TypedLocality Local]++instance forall loc a. CanRename (ComputeNode loc a) String where+  -- There is no need to update the id, as this field is not involved+  -- in the calculation of the id.+  -- TODO: make this fail immediately? If the name is wrong, it is+  -- harder to figure out what is happening.+  (@@) cn name = cn { _cnName = Just nn } where+    nn = NodeName . T.pack $ name
+ src/Spark/Core/Internal/FunctionsInternals.hs view
@@ -0,0 +1,239 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FunctionalDependencies #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE ScopedTypeVariables #-}+-- A number of utilities related to data sets and dataframes.++module Spark.Core.Internal.FunctionsInternals(+  DynColPackable,+  StaticColPackable2,+  NameTuple(..),+  TupleEquivalence(..),+  asCol,+  pack1,+  pack,+  pack',+  struct',+  struct,+) where++import Control.Arrow+import qualified Data.Vector as V+import qualified Data.Text as T+import Data.List(sort, nub)+import Formatting++import Spark.Core.Internal.ColumnStructures+import Spark.Core.Internal.ColumnFunctions+import Spark.Core.Internal.DatasetFunctions+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.Utilities+import Spark.Core.Internal.TypesGenerics+import Spark.Core.Internal.TypesFunctions+import Spark.Core.Internal.TypesStructures+import Spark.Core.Internal.OpStructures+import Spark.Core.StructuresInternal+import Spark.Core.Try++{-| The class of pairs of types that express the fact that some type a can+be converted to a dataset of type b.++This class is only inhabited by some internal types: lists, tuples, etc.+-}+class DynColPackable a where+  -- Returns (possibly) some form of the type a packed into a single column.+  -- This implementation must make sure that the final column is either a+  -- failure or is well-formed (no name duplicates, etc.)+  _packAsColumn :: a -> DynColumn++{-| The class of pairs of types that express the fact that some type a can+be converted to a dataset of type b.++This class is meant to be extended by users to create converters associated+to their data types.+-}+class StaticColPackable2 ref a b | a -> ref where+  _staticPackAsColumn2 :: a -> Column ref b++data NameTuple to = NameTuple [String]++{-| A class that expresses the fact that a certain type (that is well-formed)+is equivalent to a tuple of points.++Useful for auto conversions between tuples of columns and data structures.+-}+class TupleEquivalence to tup | to -> tup where+  tupleFieldNames :: NameTuple to++-- Here is the basic algorithm:+--  - datasets can only contain rows of things+--  - columns and observables contain cells (which may be empty)+--  - a strict struct cell is equivalent to a row+--  - a non-strict or non-struct cell is equivalent to a row with a single item+--  - as a consequence, there is no "row with a unique field". This is equivalent+--    to the element inside the field++-- Invariants to respect in terms of types (not in terms of values)+--   untypedCol . asCol == asCol'+--   pack1 . asCol == asCol . pack1+--   for single columns, pack = Right . pack1++-- The typed function+-- This only works for inner types that are known to the Haskell type system+-- fun :: (SQLTypeable a, HasCallStack) => Column a -> Column a -> Column a+-- fun = undefined++-- The untyped equivalent+-- Each of the inputs can be either a column or a try, and the final outcome is always a try+-- When both types are known to the type system, the 2 calls are equivalent+-- fun' :: (ColumnLike a1, ColumnLike a2, HasCallStack) => a1 -> a2 -> Try DynColumn+-- fun' = undefined++-- | Represents a dataframe as a single column.+asCol :: (HasCallStack) => Dataset a -> Column a a+asCol ds =+  -- Simply recast the dataset as a column.+  -- The empty path indicates that we are wrapping the whole thing.+  iEmptyCol ds (unsafeCastType $ nodeType ds) (FieldPath V.empty)++-- | Packs a single column into a dataframe.+pack1 :: (HasCallStack) => Column ref a -> Dataset a+pack1 c =+  emptyDataset (NodeStructuredTransform (colOp c)) (colType c)+      `parents` [untyped (colOrigin c)]++{-| Packs a number of columns into a single dataframe.++This operation is checked for same origin and no duplication of columns.++This function accepts columns, list of columns and tuples of columns (both+typed and untyped).+-}+pack' :: (DynColPackable a) => a -> DataFrame+-- Pack the columns and check that they have the same origin.+pack' z = pack1 <$> _packAsColumn z++{-| Packs a number of columns with the same references into a single dataset.++The type of the dataset must be provided in order to have proper type inference.++TODO: example.+-}+pack :: forall ref a b. (StaticColPackable2 ref a b, HasCallStack) => a -> Dataset b+pack z =+  let c = _staticPackAsColumn2 z :: ColumnData ref b+  in pack1 c++{-| Packs a number of columns into a single column (the struct construct).++Columns must have different names, or an error is returned.+-}+struct' :: (HasCallStack) => [DynColumn] -> DynColumn+struct' cols = do+  l <- sequence cols+  let fields = (colFieldName &&& id) <$> l+  _buildStruct fields++{-| Packs a number of columns into a single structure, given a return type.++The field names of the columns are discarded, and replaced by the field names+of the structure.+-}+struct :: forall ref a b. (StaticColPackable2 ref a b, HasCallStack) => a -> Column ref b+struct = _staticPackAsColumn2+++instance forall x. (DynColPackable x) => DynColPackable [x] where+  _packAsColumn = struct' . (_packAsColumn <$>)++instance DynColPackable DynColumn where+  _packAsColumn = id++instance forall ref a. DynColPackable (Column ref a) where+  _packAsColumn = pure . iUntypedColData++instance forall z1 z2. (DynColPackable z1, DynColPackable z2) => DynColPackable (z1, z2) where+  _packAsColumn (c1, c2) = struct' [_packAsColumn c1, _packAsColumn c2]++-- ******** Experimental ************+instance forall ref a. StaticColPackable2 ref (Column ref a) a where+  _staticPackAsColumn2 = id++-- Tuples are equivalent to tuples+instance forall a1 a2. TupleEquivalence (a1, a2) (a1, a2) where+  tupleFieldNames = NameTuple ["_1", "_2"]+++-- The equations that bind column packable stuff through their tuple equivalents+instance forall ref b a1 a2 z1 z2. (+          SQLTypeable b,+          TupleEquivalence b (a1, a2),+          StaticColPackable2 ref z1 a1,+          StaticColPackable2 ref z2 a2) =>+  StaticColPackable2 ref (z1, z2) b where+    _staticPackAsColumn2 (c1, c2) =+      let+        x1 = iUntypedColData (_staticPackAsColumn2 c1 :: Column ref a1)+        x2 = iUntypedColData (_staticPackAsColumn2 c2 :: Column ref a2)+        names = tupleFieldNames :: NameTuple b+      in _unsafeBuildStruct [x1, x2] names++instance forall ref b a1 a2 a3 z1 z2 z3. (+          SQLTypeable b,+          TupleEquivalence b (a1, a2, a3),+          StaticColPackable2 ref z1 a1,+          StaticColPackable2 ref z2 a2,+          StaticColPackable2 ref z3 a3) =>+  StaticColPackable2 ref (z1, z2, z3) b where+    _staticPackAsColumn2 (c1, c2, c3) =+      let+        x1 = iUntypedColData (_staticPackAsColumn2 c1 :: Column ref a1)+        x2 = iUntypedColData (_staticPackAsColumn2 c2 :: Column ref a2)+        x3 = iUntypedColData (_staticPackAsColumn2 c3 :: Column ref a3)+        names = tupleFieldNames :: NameTuple b+      in _unsafeBuildStruct [x1, x2, x3] names++_unsafeBuildStruct :: (HasCallStack, SQLTypeable x) =>+  [UntypedColumnData] -> NameTuple x -> Column ref x+_unsafeBuildStruct cols (NameTuple names) =+  if length cols /= length names+    then failure $ sformat ("The number of columns and names differs:"%sh%" and "%sh) cols names+    else+      let fnames = unsafeFieldName . T.pack <$> names+          uc = _buildStruct (fnames `zip` cols)+          z = forceRight uc+      in z { _cOp = _cOp z }+++_buildStruct :: [(FieldName, UntypedColumnData)] -> Try UntypedColumnData+_buildStruct [] = tryError "You cannot build an empty structure"+_buildStruct ((hfn, hcol):t) =+  let cols = ((hfn, hcol):t)+      cols' = V.fromList cols+      fields = ColStruct $ (uncurry TransformField .(fst &&& colOp . snd)) <$> cols'+      ct = StructType $ (uncurry StructField . (fst &&& unSQLType . colType . snd)) <$> cols'+      name = "struct(" <> T.intercalate "," (unFieldName . fst <$> cols) <> ")"+      names = fst <$> cols+      numNames = length names+      numDistincts = length . nub $ names+      origins = _columnOrigin (snd <$> cols)+  in case (origins, numNames == numDistincts) of+    ([_], True) ->+        pure ColumnData {+                    _cOrigin = _cOrigin hcol,+                    _cType = StrictType $ Struct ct,+                    _cOp = fields,+                    _cReferingPath = Just $ unsafeFieldName name+                  }+    (l, True) -> tryError $ sformat ("Too many distinct origins: "%sh) l+    (_, False) -> tryError $ sformat ("Duplicate field names when building the struct: "%sh) (sort names)++_columnOrigin :: [UntypedColumnData] -> [UntypedDataset]+_columnOrigin l =+  let+    groups = myGroupBy' (nodeId . colOrigin) l+  in (colOrigin . head . snd) <$> groups
+ src/Spark/Core/Internal/LocalDataFunctions.hs view
@@ -0,0 +1,101 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- A number of functions related to local data.++module Spark.Core.Internal.LocalDataFunctions(+  constant+) where++import Data.Aeson(toJSON, Value(Null))+import qualified Data.Text as T+import Control.Exception.Base(assert)++import Spark.Core.Internal.DatasetFunctions+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.TypesStructures(unSQLType)+import Spark.Core.Internal.TypesFunctions(intType)+import Spark.Core.Internal.OpStructures+import Spark.Core.Internal.Utilities+import Spark.Core.Types+import Spark.Core.Row++constant :: (ToSQL a, SQLTypeable a) => a -> LocalData a+constant cst =+  let+    sqlt = buildType+    dt = unSQLType sqlt+  in emptyLocalData (NodeLocalLit dt (toJSON (valueToCell cst))) sqlt++instance (Num a, ToSQL a, SQLTypeable a) => Num (LocalData a) where+  (+) = _binOp "org.spark.LocalPlus"+  (-) = _binOp "org.spark.LocalMinus"+  (*) = _binOp "org.spark.LocalMult"+  abs = _unaryOp "org.spark.LocalAbs"+  signum = _unaryOp "org.spark.LocalSignum"+  fromInteger x = constant (fromInteger x :: a)+  negate = _unaryOp "org.spark.LocalNegate"++instance forall a. (ToSQL a, Enum a, SQLTypeable a) => Enum (LocalData a) where+  toEnum x = constant (toEnum x :: a)+  fromEnum = failure "Cannot use fromEnum against a local data object"+  -- TODO(kps) some of the others are still available for implementation++instance (Num a, Ord a) => Ord (LocalData a) where+  compare = failure "You cannot compare instances of LocalData. (yet)."+  min = _binOp "org.spark.LocalMin"+  max = _binOp "org.spark.LocalMax"++instance forall a. (Real a, ToSQL a, SQLTypeable a) => Real (LocalData a) where+  toRational = failure "Cannot convert LocalData to rational"++instance (ToSQL a, Integral a, SQLTypeable a) => Integral (LocalData a) where+  quot = _binOp "org.spark.LocalQuotient"+  rem = _binOp "org.spark.LocalReminder"+  div = _binOp "org.spark.LocalDiv"+  mod = _binOp "org.spark.LocalMod"+  quotRem = failure "quotRem is not implemented (yet). Use quot and rem."+  divMod = failure "divMod is not implemented (yet). Use div and mod."+  toInteger = failure "Cannot convert LocalData to integer"++instance (ToSQL a, SQLTypeable a, Fractional a) => Fractional (LocalData a) where+  fromRational x = constant (fromRational x :: a)+  (/) = _binOp "org.spark.LocalDiv"+++_unaryOp :: T.Text -> LocalData a -> LocalData a+_unaryOp optxt ld =+  let so = StandardOperator {+            soName = optxt,+            soOutputType = unSQLType $ nodeType ld,+            soExtra = Null }+      op = NodeLocalOp so in+  emptyLocalData op (nodeType ld)+    `parents` [untyped ld]++_binOp :: T.Text -> LocalData a -> LocalData a -> LocalData a+_binOp optxt ld1 ld2 = assert (nodeType ld1 == nodeType ld2) $+  let so = StandardOperator {+          soName = optxt,+          soOutputType = unSQLType $ nodeType ld1,+          soExtra = Null }+      op = NodeLocalOp so in+  emptyLocalData op (nodeType ld1)+    `parents` [untyped ld1, untyped ld2]++-- TODO(kps) more input tests+_binOp' :: StandardOperator -> LocalData a -> LocalData a -> LocalData a+_binOp' so ld1 ld2 = assert (nodeType ld1 == nodeType ld2) $+  let op = NodeLocalOp so in+  emptyLocalData op (nodeType ld1)+    `parents` [untyped ld1, untyped ld2]++_intOperator :: T.Text -> StandardOperator+_intOperator optxt = StandardOperator {+  soName = optxt,+  soOutputType = intType,+  soExtra = Null+}
+ src/Spark/Core/Internal/LocatedBase.hs view
@@ -0,0 +1,41 @@+-- Taken from https://hackage.haskell.org/package/located-base-0.1.1.0/docs/src/GHC-Err-Located.html++{-# LANGUAGE CPP #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE ImplicitParams #-}+{-# LANGUAGE OverloadedStrings #-}+module Spark.Core.Internal.LocatedBase (error, undefined, HasCallStack, showCallStack) where++#if __GLASGOW_HASKELL__ < 800++import GHC.SrcLoc+import GHC.Stack (CallStack, getCallStack)+import Prelude hiding (error, undefined)+import qualified Prelude+import Text.Printf+import Data.Text(Text, unpack)++type HasCallStack = (?callStack :: CallStack)++error :: HasCallStack => Text -> a+error msg = Prelude.error (unpack msg ++ "\n" ++ showCallStack ?callStack)++undefined :: HasCallStack => a+undefined = error "Prelude.undefined"++showCallStack :: CallStack -> String+showCallStack stk = case getCallStack stk of+  _:locs -> unlines $ "Callstack:" : map format locs+  _ -> Prelude.error "showCallStack: empty call-stack"+  where+  format (fn, loc) = printf "  %s, called at %s" fn (showSrcLoc loc)++#else++import GHC.Stack++{-# DEPRECATED showCallStack "use GHC.Stack.prettyCallStack instead" #-}+showCallStack :: CallStack -> String+showCallStack = prettyCallStack++#endif
+ src/Spark/Core/Internal/OpFunctions.hs view
@@ -0,0 +1,114 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}+++module Spark.Core.Internal.OpFunctions(+  simpleShowOp,+  extraNodeOpData,+  hashUpdateNodeOp,+  prettyShowColOp,+) where++import qualified Data.Text as T+import qualified Data.Aeson as A+import qualified Data.Vector as V+import qualified Data.ByteString as BS+import qualified Data.ByteString.Lazy as LBS+import Data.Aeson((.=), toJSON)+import Data.Char(isSymbol)+import qualified Crypto.Hash.SHA256 as SHA++import Spark.Core.Internal.OpStructures+import Spark.Core.Internal.Utilities++-- (internal)+-- The serialized type of a node operation, as written in+-- the JSON description.+simpleShowOp :: NodeOp -> T.Text+simpleShowOp (NodeLocalOp op) = soName op+simpleShowOp (NodeDistributedOp op) = soName op+simpleShowOp (NodeLocalLit _ _) = T.pack "org.spark.LocalConstant"+simpleShowOp (NodeOpaqueAggregator op) = soName op+simpleShowOp (NodeUniversalAggregator ua) = soName . uaoMergeBuffer $ ua+simpleShowOp (NodeStructuredTransform _) = T.pack "org.spark.Select"+simpleShowOp (NodeDistributedLit _ _) = T.pack "org.spark.Constant"++-- A human-readable string that represents column operations.+prettyShowColOp :: ColOp -> T.Text+prettyShowColOp (ColExtraction fpath) = T.pack (show fpath)+prettyShowColOp (ColFunction txt cols) =+  _prettyShowColFun txt (V.toList cols)+prettyShowColOp (ColLit _ cell) = T.pack (show cell)+prettyShowColOp (ColStruct s) =+  "struct(" <> T.intercalate "," (prettyShowColOp . tfValue <$> V.toList s) <> ")"++-- (internal)+-- The extra data associated with the operation, and that is required+-- by the backend to successfully perform the operation.+-- We pass the type as seen by krapsh (along with some extra information about+-- nullability). This information is required by spark to analyze the exact+-- type of some operations.+extraNodeOpData :: NodeOp -> A.Value+extraNodeOpData (NodeLocalLit dt cell) =+  A.object [ "type" .= toJSON dt,+             "content" .= toJSON cell]+extraNodeOpData (NodeStructuredTransform st) = toJSON st+extraNodeOpData (NodeDistributedLit dt lst) =+  -- The backend deals with all the details translating the augmented type+  -- as a SQL datatype.+  A.object [ "cellType" .= toJSON dt,+             "content" .= toJSON lst]+extraNodeOpData _ = A.Null++-- Adds the content of a node op to a hash.+-- Right now, this builds the json representation and passes it+-- to the hash function, which simplifies the verification on+-- on the server side.+-- TODO: this depends on some implementation details such as the hashing+-- function used by Aeson.+hashUpdateNodeOp :: SHA.Ctx -> NodeOp -> SHA.Ctx+hashUpdateNodeOp ctx op = _hashUpdateJson ctx $ A.object [+  "op" .= simpleShowOp op,+  "extra" .= extraNodeOpData op]+++_prettyShowColFun :: T.Text -> [ColOp] -> T.Text+_prettyShowColFun txt [col] | _isSym txt =+  T.concat [txt, prettyShowColOp col]+_prettyShowColFun txt [col1, col2] | _isSym txt =+  -- This is not perfect for complex operations, but it should get the job done+  -- for now.+  -- TODO eventually use operator priority here+  T.concat [prettyShowColOp col1, txt, prettyShowColOp col2]+_prettyShowColFun txt cols =+  let vals = T.intercalate ", " (prettyShowColOp <$> cols) in+  T.concat [txt, "(", vals, ")"]++_isSym :: T.Text -> Bool+_isSym txt = all isSymbol (T.unpack txt)++-- This schema is not great because there is some ambiguity about the final+-- nodes.+-- Someone could craft a JSON that would confuse the object detection.+-- Not sure if this is much of a security risk anyway.+instance A.ToJSON ColOp where+  toJSON (ColExtraction fp) = A.object [+    "colOp" .= T.pack "extraction",+    "field" .= toJSON fp]+  toJSON (ColFunction txt cols) = A.object [+    "colOp" .= T.pack "fun",+    "function" .= txt,+    "args" .= (toJSON <$> cols)]+  toJSON (ColLit _ cell) = A.object [+    "colOp" .= T.pack "literal",+    "lit" .= toJSON cell]+  toJSON (ColStruct v) =+    let fun (TransformField fn colOp) =+          A.object ["name" .= T.pack (show fn), "op" .= toJSON colOp]+    in A.Array $ fun <$> v+++_hashUpdateJson :: SHA.Ctx -> A.Value -> SHA.Ctx+_hashUpdateJson ctx val = SHA.update ctx bs where+  bs = BS.concat . LBS.toChunks . encodeDeterministicPretty $ val
+ src/Spark/Core/Internal/OpStructures.hs view
@@ -0,0 +1,164 @@+{-|+A description of the operations that can be performed on+nodes and columns.+-}+module Spark.Core.Internal.OpStructures where++import Data.Text as T+import Data.Aeson(Value, Value(Null))+import Data.Vector(Vector)++import Spark.Core.StructuresInternal+import Spark.Core.Internal.TypesStructures(DataType, SQLType, SQLType(unSQLType))++{-| The invariant respected by a transform.++Depending on the value of the invariant, different optimizations+may be available.+-}+data TransformInvariant =+    -- | This operator has no special property. It may depend on+    -- the partitioning layout, the number of partitions, the order+    -- of elements in the partitions, etc.+    -- This sort of operator is unwelcome in Krapsh...+    Opaque+    -- | This operator respects the canonical partition order, but may+    -- not have the same number of elements.+    -- For example, this could be a flatMap on an RDD (filter, etc.).+    -- This operator can be used locally with the signature a -> [a]+  | PartitioningInvariant+    -- | The strongest invariant. It respects the canonical partition order+    -- and it outputs the same number of elements.+    -- This is typically a map.+    -- This operator can be used locally with the signature a -> a+  | DirectPartitioningInvariant+++-- | The dynamic value of locality.+-- There is still a tag on it, but it can be easily dropped.+data Locality =+    -- | The data associated to this node is local. It can be materialized+    -- and accessed by the user.+    Local+    -- | The data associated to this node is distributed or not accessible+    -- locally. It cannot be accessed by the user.+  | Distributed deriving (Show, Eq)++-- ********* PHYSICAL OPERATORS ***********+-- These structures declare some operations that correspond to operations found+-- in Spark itself, or in the surrounding libraries.++-- | An operator defined by default in the release of Krapsh.+-- All other physical operators can be converted to a standard operators.+data StandardOperator = StandardOperator {+  soName :: !T.Text,+  soOutputType :: !DataType,+  soExtra :: !Value+} deriving (Eq, Show)++-- | A scala method of a singleton object.+data ScalaStaticFunctionApplication = ScalaStaticFunctionApplication {+  sfaObjectName :: !T.Text,+  sfaMethodName :: !T.Text+  -- TODO add the input and output types?+}+++-- | The different kinds of column operations.+-- These operations describe the physical operations on columns as supported+-- by Spark SQL. They can operate on column -> column, column -> row, row->row.+-- Of course, not all operators are valid for each configuration.+data ColOp =+    -- | A projection onto a single column+    -- An extraction is always direct.+    ColExtraction !FieldPath+    -- | A function of other columns.+    -- In this case, the other columns may matter+    -- TODO(kps) add if this function is partition invariant.+    -- It should be the case most of the time.+  | ColFunction !T.Text !(Vector ColOp)+    -- | A constant defined for each element.+    -- The type should be the same as for the column+    -- A literal is always direct+  | ColLit !DataType !Value+    -- | A structure.+  | ColStruct !(Vector TransformField)+  deriving (Eq, Show)++-- | A field in a structure.+data TransformField = TransformField {+  tfName :: !FieldName,+  tfValue :: !ColOp+} deriving (Eq, Show)++-- | The content of a structured transform.+data StructuredTransform =+    InnerOp !ColOp+  | InnerStruct !(Vector TransformField)+  deriving (Eq, Show)+++-- ********* DATASET OPERATORS ************+-- These describe Dataset -> Dataset transforms.+++data DatasetTransformDesc =+    DSScalaStaticFunction !ScalaStaticFunctionApplication+  | DSStructuredTransform !ColOp+  | DSOperator !StandardOperator+++-- ****** OBSERVABLE OPERATORS *******+-- These operators describe Observable -> Observable transforms++-- **** AGGREGATION OPERATORS *****+-- The different types of aggregators++-- The low-level description of a+-- The name of the aggregator is the name of the+-- Dataset -> Local data transform+data UniversalAggregatorOp = UniversalAggregatorOp {+  uaoMergeType :: !DataType,+  uaoInitialOuter :: !StandardOperator,+  uaoMergeBuffer :: !StandardOperator+} deriving (Eq, Show)++{-+A node operation.+A description of all the operations between nodes.+These are the low-level, physical operations that Spark implements.++Each node operation is associated with:+ - a locality+ - an operation name (implicit or explicit)+ - a data type+ - a representation in JSON++Additionally, some operations are associated with algebraic invariants+to enable programmatic transformations.+-}+data NodeOp =+    -- | An operation between local nodes: [Observable] -> Observable+    NodeLocalOp StandardOperator+    -- | An observable literal+  | NodeLocalLit !DataType !Value+    -- | Some aggregator that does not respect any particular invariant.+  | NodeOpaqueAggregator StandardOperator+    -- | A universal aggregator.+  | NodeUniversalAggregator UniversalAggregatorOp+    -- | A structured transform, performed either on a local node or a+    -- distributed node.+  | NodeStructuredTransform !ColOp+    -- | A distributed dataset (with no partition information)+  | NodeDistributedLit !DataType !(Vector Value)+    -- | An opaque distributed operator.+  | NodeDistributedOp StandardOperator+  deriving (Eq, Show)++-- | Makes a standard operator with no extra value+makeOperator :: T.Text -> SQLType a -> StandardOperator+makeOperator txt sqlt =+  StandardOperator {+    soName = txt,+    soOutputType = unSQLType sqlt,+    soExtra = Null }
+ src/Spark/Core/Internal/Paths.hs view
@@ -0,0 +1,186 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FlexibleContexts #-}++module Spark.Core.Internal.Paths(+  HasNodeName(..),+  PathEdge(..),+  computePaths,+  assignPaths',+  -- For testing:+  Scopes,+  ParentSplit(..),+  mergeScopes,+  gatherPaths,+  iGetScopes0,+) where++import qualified Data.Map.Strict as M+import qualified Data.Set as S+import qualified Data.Vector as V+import Data.List(sort)+import Data.Maybe(fromMaybe, catMaybes)+import Data.Foldable(foldr', foldl', toList)+import Formatting++import Spark.Core.Try+import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.Utilities+import Spark.Core.Internal.ComputeDag+import Spark.Core.StructuresInternal++class HasNodeName v where+  -- Retrieves the name of the node+  getNodeName :: v -> NodeName+  -- Assigns a path to the node+  assignPath :: v -> NodePath -> v++{-| The types of edges for the calculation of paths.+ - same level parent -> the node should have the same prefix as its parents+ - inner edge -> the parent defines the scope of this node+ -}+data PathEdge = SameLevelEdge | InnerEdge deriving (Show, Eq)++-- Assigns paths in a graph.+--+computePaths :: (HasCallStack, HasNodeName v) =>+  ComputeDag v PathEdge -> Try (M.Map VertexId NodePath)+computePaths cd =+  let nodecg = mapVertexData getNodeName cd+  in _computePaths nodecg++assignPaths' :: (HasCallStack, HasNodeName v) =>+  M.Map VertexId NodePath -> ComputeDag v e -> ComputeDag v e+assignPaths' m cd =+  let f vx =+        let old = NodePath . V.singleton $ getNodeName (vertexData vx)+            new = M.findWithDefault old (vertexId vx) m+        in assignPath (vertexData vx) new+  in mapVertices f cd++-- The main function to perform the pass assignments.+-- It starts from the graph of dependencies and from the local name info,+-- and computes the complete paths (if possible), starting from the fringe.+_computePaths :: (HasCallStack) =>+  ComputeDag NodeName PathEdge -> Try (M.Map VertexId NodePath)+_computePaths cg =+  let+    scopes = iGetScopes0 (toList . cdOutputs $ cg) (_splitParents' (cdEdges cg))+    paths = gatherPaths scopes+    nodeNames = M.fromList [(vertexId vx, vertexData vx)| vx <- toList . cdVertices $ cg]+    lookup' nid = M.lookup nid nodeNames+    f :: VertexId -> [[VertexId]] -> Try NodePath+    f nid ls = case ls of+      [l] ->+        return . NodePath . V.fromList . catMaybes $ lookup' <$> (l ++ [nid])+      x ->+        tryError $ sformat ("Node has too many paths: node="%shown%" discovered paths ="%shown) nid x+    nodePaths = M.traverseWithKey f paths+  in nodePaths++-- (private)+-- The top-level scope may not have an ID associated to it+type Scopes = M.Map (Maybe VertexId) (S.Set VertexId)+++-- (internal)+-- The separation of parents into logical and inner parents+data ParentSplit a = ParentSplit {+  psLogical :: ![Vertex a],+  psInner :: ![Vertex a]+} deriving (Show)++_lookupOrEmpty :: Scopes -> Maybe VertexId -> [VertexId]+_lookupOrEmpty scopes mnid =+  S.toList $ fromMaybe S.empty (M.lookup mnid scopes)++mergeScopes :: Scopes -> Scopes -> Scopes+mergeScopes = M.unionWith S.union++_singleScope :: Maybe VertexId -> VertexId -> Scopes+_singleScope mKey nid = M.singleton mKey (S.singleton nid)++-- For each node, finds the one, or more than one if possible, path(s)+-- from the root to the node (which is itself not included at the end)+-- The gathering of paths may not be exaustive.+gatherPaths :: Scopes -> M.Map VertexId [[VertexId]]+gatherPaths scopes = M.map sort $ _gatherPaths0 scopes start where+  start = _lookupOrEmpty scopes Nothing++_gatherPaths0 :: Scopes -> [VertexId] -> M.Map VertexId [[VertexId]]+_gatherPaths0 _ [] = M.empty+_gatherPaths0 scopes (nid : t) =+  let+    inner = _lookupOrEmpty scopes (Just nid)+    innerPaths = _gatherPaths0 scopes inner+    innerWithHead = M.map (\l -> (nid : ) <$> l) innerPaths+    thisPaths = M.singleton nid [[]]+    innerPaths2 = M.unionWith (++) innerWithHead thisPaths+  in M.unionWith (++) innerPaths2 (_gatherPaths0 scopes t)+++iGetScopes0 :: forall a. (Show a) =>+  [Vertex a] ->+  (Vertex a -> ParentSplit a) ->+  Scopes+iGetScopes0 [] _splitter = M.empty+iGetScopes0 (h : t) splitter =+  let+    startScope = _singleScope Nothing (vertexId h)+    folder :: Scopes -> Vertex a -> Scopes+    folder current un =+      if M.member (Just (vertexId un)) current then+        current+      else+        let split = _getScopes' splitter Nothing S.empty un current+        in mergeScopes split current+  -- Important here to use a left folder, as we want to start with the head+  -- and move down the list.+  in foldl' folder startScope (h : t)++_splitParents' :: AdjacencyMap v PathEdge -> Vertex v -> ParentSplit v+_splitParents' m vx =+  let ves = V.toList $ M.findWithDefault V.empty (vertexId vx) m+      scope = [veEndVertex ve | ve <- ves, edgeData (veEdge ve) == SameLevelEdge]+      parents' = [veEndVertex ve | ve <- ves, edgeData (veEdge ve) == InnerEdge]+  in ParentSplit { psLogical = scope, psInner = parents' }+++-- TODO(kps) this recursive code is most probably going to explode for deep stacks+_getScopes' :: forall a. (Show a) =>+  (Vertex a -> ParentSplit a) -> -- The expansion of a node into logical and inner nodes+  Maybe VertexId -> -- the current parent (if any)+  S.Set VertexId -> -- the current boundary to respect+  Vertex a -> -- the current node to expand+  Scopes -> -- the scopes seen so far+  Scopes+_getScopes' splitter mScopeId boundary un scopes =+  if S.member (vertexId un) boundary then+    scopes+  else+    let+      split = splitter un+      logParents = psLogical split+      innerParents = psInner split+      -- A fold on the parents+      parF :: Vertex a -> Scopes -> Scopes+      parF v s =+        -- Same boundary and parent, but update the scopes+        _getScopes' splitter mScopeId boundary v s+      scopesPar = foldr' parF scopes logParents+      -- Now work on the inner nodes:+      vid = vertexId un+      boundary' = S.fromList (vertexId <$> logParents)+      inF :: Vertex a -> Scopes -> Scopes+      inF v s =+        -- parent is current, boundary is current logical+        _getScopes' splitter (Just vid) boundary' v s+      scopesIn = foldr' inF scopesPar innerParents+      scopesFinal = scopesIn+          `mergeScopes` _singleScope mScopeId vid+          `mergeScopes` M.singleton (Just vid) S.empty+    in scopesFinal
+ src/Spark/Core/Internal/PathsUntyped.hs view
@@ -0,0 +1,118 @@+{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE TypeSynonymInstances #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.Internal.PathsUntyped(+  assignPathsUntyped,+  tieNodes+) where++import qualified Data.Vector as V+import qualified Data.Map.Strict as M+import Data.Maybe(fromMaybe)+import Data.Foldable(toList)+import Data.List(nub)+import Control.Arrow((&&&))+import Formatting+import Control.Monad.Identity++import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.DAGFunctions+import Spark.Core.Internal.ComputeDag+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.DatasetFunctions+import Spark.Core.Internal.Paths+import Spark.Core.Internal.Utilities+import Spark.Core.Try+import Spark.Core.StructuresInternal(unNodeId)++instance GraphVertexOperations UntypedNode where+  vertexToId = VertexId . unNodeId . nodeId+  expandVertexAsVertices n =+    nodeParents n+      ++ fromMaybe [] (nodeLogicalParents n)+      ++ nodeLogicalDependencies n++instance GraphOperations UntypedNode NodeEdge where+  expandVertex n =+    -- The logical parents are more important than the parents+    let+      -- If the logical parents are not specified, the logical parents are the+      -- direct parents.+      scopeNodes = fromMaybe (nodeParents n) (nodeLogicalParents n)+      loParents = [(ScopeEdge, v) | v <- scopeNodes]+      -- The direct parents. They may overload with the scoping parents, but+      -- this will be checked during the name analysis.+      parents' = (const (DataStructureEdge ParentEdge) &&& id) <$> nodeParents n+      loDeps = (const (DataStructureEdge LogicalEdge) &&& id) <$> nodeLogicalDependencies n+    in loParents ++ parents' ++ loDeps++instance HasNodeName UntypedNode where+  getNodeName = nodeName+  assignPath n p = updateNode n $ \n' -> n' { _cnPath = p }+++-- Stitches the nodes together to make sure that the edges in the graph also+-- correspond to the dependencies in the nodes themselves.+-- This does not update the nodeIds+-- TODO is it needed?+tieNodes :: ComputeDag UntypedNode StructureEdge -> ComputeDag UntypedNode StructureEdge+tieNodes cd =+  let g = computeGraphToGraph cd+      f :: UntypedNode -> [(UntypedNode, StructureEdge)] -> Identity UntypedNode+      f v l =+        let parents' = V.fromList [n | (n, e) <- l, e == ParentEdge]+            logDeps = V.fromList [n | (n, e) <- l, e == LogicalEdge]+            res = updateNode v $ \n -> n {+                  _cnParents = parents',+                  _cnLogicalDeps = logDeps,+                  _cnLogicalParents = Nothing }+        in return res+      g2 = runIdentity $ graphMapVertices g f+  in graphToComputeGraph g2++-- Assigs the paths, and drops the scoping edges.+assignPathsUntyped :: (HasCallStack) =>+  ComputeDag UntypedNode NodeEdge -> Try (ComputeDag UntypedNode StructureEdge)+assignPathsUntyped cd = do+  let pathCGraph = _getPathCDag cd+  paths <- computePaths pathCGraph+  let g = computeGraphToGraph $ assignPaths' paths cd+  let f ScopeEdge = []+      f (DataStructureEdge x) = [x]+  let g' = graphFlatMapEdges g  f+  return $ graphToComputeGraph g'+++-- transforms node edges into path edges+_cleanEdges :: (HasCallStack) => [VertexEdge NodeEdge v] -> [VertexEdge PathEdge v]+_cleanEdges [] = []+_cleanEdges (h : t) =+  let vid = vertexId (veEndVertex h)+      others = [ve | ve <- t, (vertexId . veEndVertex $ ve) /= vid]+      sames = [ve | ve <- t, (vertexId . veEndVertex $ ve) == vid]+      rest = _cleanEdges others+      e = veEdge h+      -- If there multiple edges between nodes, they are dropped.+      -- This distinction is not required for names.+      eData = nub $ edgeData . veEdge <$> (h : sames)+      eData' = case eData of+        [DataStructureEdge ParentEdge] -> Just InnerEdge+        [DataStructureEdge ParentEdge, ScopeEdge] -> Just SameLevelEdge+        [ScopeEdge, DataStructureEdge ParentEdge] -> Just SameLevelEdge+        [ScopeEdge] -> Just SameLevelEdge+        [DataStructureEdge LogicalEdge] -> Nothing+        l -> failure (sformat ("Could not understand combination "%shown) l)+      res = case eData' of+        Just v -> (h { veEdge = e { edgeData = v } }) : rest+        Nothing -> rest+    in res+++_getPathCDag :: (HasCallStack) => ComputeDag v NodeEdge -> ComputeDag v PathEdge+_getPathCDag cd =+  let adj' = M.map (V.fromList . _cleanEdges . toList) (cdEdges cd)+  in cd { cdEdges = adj' }
+ src/Spark/Core/Internal/RowGenerics.hs view
@@ -0,0 +1,96 @@+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE MultiParamTypeClasses #-}+++-- The generic implementation for the protocol that converts to+-- and from SQL cells.+-- Going through JSON is not recommended because of precision loss+-- for the numbers, and other issues related to numbers.+module Spark.Core.Internal.RowGenerics(+  ToSQL,+  valueToCell,+) where++import GHC.Generics+import qualified Data.Vector as V+import Data.Text(pack)++import Spark.Core.Internal.RowStructures+import Spark.Core.Internal.Utilities++-- We need to differentiate between the list built for the+-- constructor and an inner object.+data CurrentBuffer =+  ConsData ![Cell]+  | BuiltCell !Cell deriving (Show)++_cellOrError :: CurrentBuffer -> Cell+_cellOrError (BuiltCell cell) = cell+_cellOrError x = let msg = "Expected built cell, received " ++ show x in+  failure (pack msg)++-- All the types that can be converted to a SQL value.+class ToSQL a where+  _valueToCell :: a -> Cell++  default _valueToCell :: (Generic a, GToSQL (Rep a)) => a -> Cell+  _valueToCell !x = _g2cell (from x)++valueToCell :: (ToSQL a) => a -> Cell+valueToCell = _valueToCell++-- class FromSQL a where+--   _cellToValue :: Cell -> Try a++instance ToSQL a => ToSQL (Maybe a) where+  _valueToCell (Just x) = _valueToCell x+  _valueToCell Nothing = Empty++instance ToSQL Int where+  _valueToCell = IntElement++class GToSQL r where+  _g2buffer :: r a -> CurrentBuffer+  _g2cell :: r a -> Cell+  _g2cell = _cellOrError . _g2buffer++instance GToSQL U1 where+  _g2buffer U1 = failure $ pack "GToSQL UI called"++-- | Constants, additional parameters and recursion of kind *+instance (GToSQL a, GToSQL b) => GToSQL (a :*: b) where+  _g2buffer (a :*: b) = case (_g2buffer a, _g2buffer b) of+    (ConsData l1, ConsData l2) -> ConsData (l1 ++ l2)+    (y1, y2) -> failure $ pack $ "GToSQL (a :*: b): Expected buffers, received " ++ show y1 ++ " and " ++ show y2++instance (GToSQL a, GToSQL b) => GToSQL (a :+: b) where+  _g2buffer (L1 x) = _g2buffer x+  _g2buffer (R1 x) = let !y = _g2buffer x in y++-- -- | Sums: encode choice between constructors+-- instance (GToSQL a) => GToSQL (M1 i c a) where+--   _g2cell !(M1 x) = let !y = _g2cell x in+--     trace ("GToSQL M1: y = " ++ show y) y++instance (GToSQL a, Constructor c) => GToSQL (M1 C c a) where+  _g2buffer (M1 x) = let !y = _g2buffer x in y++instance (GToSQL a, Selector c) => GToSQL (M1 S c a) where+  _g2buffer (M1 x) = let !y = ConsData [_g2cell x] in y++instance (GToSQL a, Datatype c) => GToSQL (M1 D c a) where+  _g2buffer (M1 x) =+    case _g2buffer x of+      ConsData cs -> BuiltCell $ RowArray (V.fromList cs)+      BuiltCell cell -> BuiltCell cell++-- | Products: encode multiple arguments to constructors+instance (ToSQL a) => GToSQL (K1 i a) where+  _g2buffer (K1 x) = let !y = _valueToCell x in BuiltCell y
+ src/Spark/Core/Internal/RowGenericsFrom.hs view
@@ -0,0 +1,130 @@+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+++-- The generic implementation for the protocol that converts to+-- and from SQL cells.+-- Going through JSON is not recommended because of precision loss+-- for the numbers, and other issues related to numbers.+module Spark.Core.Internal.RowGenericsFrom(+  FromSQL,+  cellToValue,+) where++import GHC.Generics+import Data.Text(Text, pack)+import Control.Monad.Except+import Formatting+import qualified Data.Vector as V++import Spark.Core.Internal.RowStructures+import Spark.Core.Internal.Utilities++-- Convert a cell to a value (if possible)+cellToValue :: (FromSQL a) => Cell -> Either Text a+cellToValue = _cellToValue++type TryS = Either Text++-- Switch between the constructor list or parsing a regular cell.+data Decode2 =+    D2Cons ![Cell]+  | D2Normal !Cell+  deriving (Eq, Show)++-- All the types that can be converted to a SQL value.+class FromSQL a where+  _cellToValue :: Cell -> TryS a++  default _cellToValue :: (Generic a, GFromSQL (Rep a)) => Cell -> TryS a+  _cellToValue cell = let+      x1r = _gFcell (D2Normal cell) :: TryS (Rep a a)+      x1t = to <$> x1r+    in x1t+++instance FromSQL a => FromSQL (Maybe a) where+  _cellToValue Empty = pure Nothing+  _cellToValue x = pure <$> _cellToValue x++instance FromSQL Int where+  _cellToValue (IntElement x) = pure x+  _cellToValue x = throwError $ sformat ("FromSQL: Decoding an int from "%shown) x++instance FromSQL Cell where+  _cellToValue = pure++instance FromSQL a => FromSQL [a] where+  _cellToValue (RowArray xs) = sequence (_cellToValue <$> V.toList xs)+  _cellToValue x = throwError $ sformat ("FromSQL: Decoding array from "%shown) x+-- ******* GENERIC ********++class GFromSQL r where+  _gFcell :: Decode2 -> TryS (r a)++instance GFromSQL U1 where+  _gFcell x = failure $ pack $ "GFromSQL UI called" ++ show x++instance (GFromSQL a, GFromSQL b) => GFromSQL (a :*: b) where+  _gFcell (D2Normal (RowArray arr)) | not (V.null arr) =+    let (cell : l) = V.toList arr+        x1t = _gFcell (D2Normal cell)+        x2t = _gFcell (D2Cons l)+        x = do+          x1 <- x1t+          x2 <- x2t+          return (x1 :*: x2)+    in x+  _gFcell (D2Cons (cell : l)) =+    let x1t = _gFcell (D2Cons [cell])+        x2t = _gFcell (D2Cons l)+        x = do+          x1 <- x1t+          x2 <- x2t+          return (x1 :*: x2)+    in x+  _gFcell x = failure $ pack ("GFromSQL (a :*: b) " ++ show x)+++instance (GFromSQL a, GFromSQL b) => GFromSQL (a :+: b) where+  _gFcell x = failure $ pack $ "GFromSQL (a :+: b)" ++ show x++_m1 :: GFromSQL f1 => String -> Decode2 -> TryS (M1 i1 c1 f1 p1)+_m1 msg (D2Cons x) = failure $ pack (msg ++ " FAILED CONS: " ++ show x)+_m1 _ (D2Normal cell) =+    let xt = _gFcell (D2Normal cell) in+      M1 <$> xt++instance (GFromSQL a, Constructor c) => GFromSQL (M1 C c a) where+  _gFcell = _m1 "GFromSQL (M1 C c a)"++instance (GFromSQL a, Selector c) => GFromSQL (M1 S c a) where+  _gFcell (D2Normal (RowArray arr)) | V.length arr == 1 =+    M1 <$> _gFcell (D2Cons [cell]) where+      cell = V.head arr+  _gFcell z @ (D2Cons [_]) =+    M1 <$> _gFcell z+  _gFcell x = _m1 "GFromSQL (M1 S c a)" x++instance (GFromSQL a, Datatype c) => GFromSQL (M1 D c a) where+  _gFcell z @ (D2Normal (RowArray _)) =+    let xt = _gFcell z in+      M1 <$> xt+  _gFcell x = failure $ pack $ "FAIL GFromSQL (M1 D c a)" ++ show x++-- | Products: encode multiple arguments to constructors+instance (FromSQL a) => GFromSQL (K1 i a) where+  _gFcell (D2Normal cell) =+    let xt = _cellToValue cell :: TryS a in+      K1 <$> xt+  _gFcell (D2Cons [cell]) =+    let xt = _cellToValue cell in+      K1 <$> xt+  _gFcell x = failure $ pack ("GFromSQLK FAIL " ++ show x)
+ src/Spark/Core/Internal/RowStructures.hs view
@@ -0,0 +1,46 @@+module Spark.Core.Internal.RowStructures where++import Data.Aeson+import Data.Vector(Vector)+import qualified Data.Text as T++-- | The basic representation of one row of data. This is a standard type that comes out of the+-- SQL engine in Spark.++-- | An element in a Row object.+-- All objects manipulated by the Spark framework are assumed to+-- be convertible to cells.+--+-- This is usually handled by generic transforms.+data Cell =+    Empty -- To represent maybe+    | IntElement !Int+    | StringElement !T.Text+    | RowArray !(Vector Cell) deriving (Show, Eq)++-- | A Row of data: the basic data structure to transport information+-- TODO rename to rowCells+data Row = Row {+    cells :: !(Vector Cell)+  } deriving (Show, Eq)+++-- AESON INSTANCES++-- TODO(kps) add some workaround to account for the restriction of+-- JSON types:+-- int32 -> int32+-- double -> double+-- weird double -> string?+-- long/bigint -> string?++-- | Cell+instance ToJSON Cell where+  toJSON Empty = Null+  toJSON (IntElement i) = toJSON i+  toJSON (StringElement s) = toJSON s+  toJSON (RowArray arr) = toJSON arr++-- | Row+instance ToJSON Row where+  toJSON (Row x) = toJSON x
+ src/Spark/Core/Internal/RowUtils.hs view
@@ -0,0 +1,112 @@+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.Internal.RowUtils(+  jsonToCell,+  checkCell,+) where++import Data.Aeson+import Data.Text(Text)+import Data.Maybe(catMaybes, listToMaybe)+import Formatting+import qualified Data.Vector as V+import qualified Data.HashMap.Strict as HM+import Data.Scientific(floatingOrInteger)+import Control.Monad.Except++import Spark.Core.Internal.TypesStructures+import Spark.Core.Internal.TypesFunctions+import Spark.Core.Internal.RowStructures+import Spark.Core.StructuresInternal(FieldName(..))+import Spark.Core.Internal.Utilities+++-- | Decodes a JSON into a row.+-- This operation requires a SQL type that describes+ -- the schema.+jsonToCell :: DataType -> Value -> Either Text Cell+jsonToCell dt v = _j2Cell v dt++{-| Given a datatype, ensures that the cell has the corresponding type.+-}+checkCell :: DataType -> Cell -> Either Text Cell+checkCell dt c = case _checkCell dt c of+  Nothing -> pure c+  Just txt -> throwError txt+++type TryCell = Either Text Cell++-- Returns an error message if something wrong is found+_checkCell :: DataType -> Cell -> Maybe Text+_checkCell dt c = case (dt, c) of+  (NullableType _, Empty) -> Nothing+  (StrictType _, Empty) ->+    pure $ sformat ("Expected a strict value of type "%sh%" but no value") dt+  (StrictType sdt, x) -> _checkCell' sdt x+  (NullableType sdt, x) -> _checkCell' sdt x++-- Returns an error message if something wrong is found+_checkCell' :: StrictDataType -> Cell -> Maybe Text+_checkCell' sdt c = case (sdt, c) of+  (_, Empty) ->+    pure $ sformat ("Expected a strict value of type "%sh%" but no value") sdt+  (IntType, IntElement _) -> Nothing+  (StringType, StringElement _) -> Nothing+  (Struct (StructType fields), RowArray cells') ->+    if V.length fields == V.length cells'+      then+        let types = V.toList $ structFieldType <$> fields+            res = uncurry _checkCell <$> (types `zip` V.toList cells')+        in listToMaybe (catMaybes res)+      else+        pure $ sformat ("Struct "%sh%" has "%sh%" fields, asked to be matched with "%sh%" cells") sdt (V.length fields) (V.length cells')+  (ArrayType dt, RowArray cells') ->+    let res = uncurry _checkCell <$> (repeat dt `zip` V.toList cells')+    in listToMaybe (catMaybes res)+  (_, _) ->+    pure $ sformat ("Type "%sh%" is incompatible with cell content "%sh) sdt c+++_j2Cell :: Value -> DataType -> TryCell+_j2Cell Null (StrictType t) =+  throwError $ sformat ("_j2Cell: Expected "%shown%", got null") t+_j2Cell Null (NullableType _) = pure Empty+_j2Cell x (StrictType t) = _j2CellS x t+-- We do not express optional types at cell level. They have to be+-- encoded in the data type.+_j2Cell x (NullableType t) = _j2CellS x t+--_j2Cell x t = throwError $ sformat ("_j2Cell: Could not match value "%shown%" with type "%shown) x t++_j2CellS :: Value -> StrictDataType -> TryCell+_j2CellS (String t) StringType = Right . StringElement $ t+_j2CellS (Array v) (ArrayType t) =+  let trys = flip _j2Cell t <$> v in+    RowArray <$> sequence trys+_j2CellS (Number s) IntType = case floatingOrInteger s :: Either Double Int of+  Left _ -> throwError $ sformat ("_j2CellS: Could not cast as int "%shown) s+  Right i -> pure (IntElement i)+-- Normal representation as object.+_j2CellS (Object o) (Struct struct) =+  let+    o2f :: StructField -> TryCell+    o2f field =+      let nullable = isNullable $ structFieldType field+          val = HM.lookup (unFieldName $ structFieldName field) o in+      case val of+        Nothing ->+          if nullable then+            pure Empty+          else throwError $ sformat ("_j2CellS: Could not find key "%shown%" in object "%shown) field o+        Just x -> _j2Cell x (structFieldType field)+    fields = o2f <$> structFields struct+  in RowArray <$> sequence fields+-- Compact array-based representation.+_j2CellS (Array v) (Struct (StructType fields)) =+  if V.length v == V.length fields+    then+      let dts = structFieldType <$> fields+          inner = uncurry _j2Cell <$> V.zip v dts+      in RowArray <$> sequence inner+    else throwError $ sformat ("_j2CellS: Compact object format a different number of fields '"%shown%"' compared "%shown) v fields+_j2CellS x t = throwError $ sformat ("_j2CellS: Could not match value '"%shown%"' with type "%shown) x t
+ src/Spark/Core/Internal/TypesFunctions.hs view
@@ -0,0 +1,107 @@+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.Internal.TypesFunctions(+  isNullable,+  iInnerStrictType,+  columnType,+  unsafeCastType,+  intType,+  arrayType,+  arrayType',+  frameTypeFromCol,+  colTypeFromFrame,+  canNull,+  structField,+  structType,+  iSingleField,+  -- cellType,+) where++import Data.Text as T+import qualified Data.Vector as V++import Spark.Core.Internal.TypesStructures+import Spark.Core.StructuresInternal++-- Performs a cast of the type.+-- This may throw an error if the required type b is not+-- compatible with the type embedded in a.+unsafeCastType :: SQLType a -> SQLType b+-- TODO add more error checking here.+unsafeCastType (SQLType dt) = SQLType dt+++-- Given a sql type tag, returns the equivalent data type for a column or a blob+-- (internal)+columnType :: SQLType a -> DataType+columnType (SQLType dt) = dt++-- (internal)+isNullable :: DataType -> Bool+isNullable (StrictType _) = False+isNullable (NullableType _) = True+++-- *** Creation of data types ***+++-- Takes a data type (assumed to be that of a column or cell) and returns the+-- corresponding dataset type.+-- This should only be used when talking to Spark.+-- All visible operations in Krapsh use Cell types instead.+-- TODO should it use value or _1? Both seem to be used in Spark.+frameTypeFromCol :: DataType -> StructType+frameTypeFromCol (StrictType (Struct struct)) = struct+frameTypeFromCol dt = _structFromUnfields [("value", dt)]++-- Given the structural type for a dataframe or a dataset, returns the+-- equivalent column type.+colTypeFromFrame :: StructType -> DataType+colTypeFromFrame st @ (StructType fs) = case V.toList fs of+  [StructField {+    structFieldName = fname,+    structFieldType = (StrictType dt)}] | fname == "value" ->+      StrictType dt+  _ -> StrictType (Struct st)+++-- The strict int type+intType :: DataType+intType = StrictType IntType++-- a string+structField :: T.Text -> DataType -> StructField+structField txt = StructField (FieldName txt)++-- The strict structure type+structType :: [StructField] -> DataType+structType = StrictType . Struct . StructType . V.fromList++-- The strict array type+arrayType' :: DataType -> DataType+arrayType' = StrictType . ArrayType++-- Returns the equivalent data type that may be nulled.+canNull :: DataType -> DataType+canNull = NullableType . iInnerStrictType++-- Given a type, returns the corresponding array type.+-- This is preferred to using directly buildType, as it may encounter some+-- overlapping instances.+arrayType :: SQLType a -> SQLType [a]+arrayType (SQLType dt) = SQLType (arrayType' dt)++iInnerStrictType :: DataType -> StrictDataType+iInnerStrictType (StrictType st) = st+iInnerStrictType (NullableType st) = st++iSingleField :: DataType -> Maybe DataType+iSingleField (StrictType (Struct (StructType fields))) = case V.toList fields of+  [StructField _ dt] -> Just dt+  _ -> Nothing+iSingleField _ = Nothing+++_structFromUnfields :: [(T.Text, DataType)] -> StructType+_structFromUnfields l = StructType . V.fromList $ x where+   x = [StructField (FieldName name) dt | (name, dt) <- l]
+ src/Spark/Core/Internal/TypesGenerics.hs view
@@ -0,0 +1,158 @@+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE DefaultSignatures #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE MultiParamTypeClasses #-}++module Spark.Core.Internal.TypesGenerics where++import qualified Data.Vector as V+import Data.Text(Text, pack)+import Data.Proxy+import GHC.Generics+import Formatting+import Debug.Trace++import Spark.Core.Internal.TypesStructures+import Spark.Core.Internal.TypesFunctions+import Spark.Core.Internal.Utilities+import Spark.Core.StructuresInternal(FieldName(..), unsafeFieldName)++-- The 3rd attempt to get generics conversions.++-- Given a tag on a type, returns the equivalent SQL type.+-- This is the type for a cell, not for a row.+-- TODO(kps) more documentation+buildType :: (SQLTypeable a) => SQLType a+buildType = _buildType+++-- The class of all the types for which the SQL type can be inferred+-- from the Haskell type only.+-- Two notable exceptions are Row and Cell, which are the dynamic types+-- used by Spark.+-- See also buildType on how to use it.+class SQLTypeable a where+  _genericTypeFromValue :: a -> GenericType+  default _genericTypeFromValue :: (Generic a, GenSQLTypeable (Rep a)) => a -> GenericType+  _genericTypeFromValue _ = genBuildType (Proxy :: Proxy a)++  -- | The only function that should matter for users in this file.+  -- Given a type, returns the SQL representation of this type.+  _buildType :: SQLType a+  _buildType =+    let !dt = _genericTypeFromValue (undefined :: a)+        SQLType u = dt in SQLType u++-- These a private types that should not be used elsewhere.+data GenericRow+type GenericType = SQLType GenericRow+++-- Generic building type.+genBuildType :: forall a. (Generic a, GenSQLTypeable (Rep a)) => Proxy a -> GenericType+genBuildType _ = genTypeFromProxy (Proxy :: Proxy (Rep a))+++instance SQLTypeable Int where+  _genericTypeFromValue _ = SQLType (StrictType IntType)++instance SQLTypeable Text where+  _genericTypeFromValue _ = SQLType (StrictType StringType)++instance {-# INCOHERENT #-} SQLTypeable String where+  _genericTypeFromValue _ = SQLType (StrictType StringType)++instance SQLTypeable a => SQLTypeable (Maybe a) where+  _genericTypeFromValue _ = let SQLType dt = buildType :: (SQLType a) in+    (SQLType . NullableType . iInnerStrictType) dt++instance {-# OVERLAPPABLE #-} SQLTypeable a => SQLTypeable [a] where+  _genericTypeFromValue _ =+    let SQLType dt = buildType :: (SQLType a) in+      (SQLType . StrictType . ArrayType) dt++instance forall a1 a2. (+    SQLTypeable a2,+    SQLTypeable a1) => SQLTypeable (a1, a2) where+  _genericTypeFromValue _ =+    let+      SQLType t1 = buildType :: SQLType a1+      SQLType t2 = buildType :: SQLType a2+    in _buildTupleStruct [t1, t2]++_buildTupleStruct :: [DataType] -> SQLType x+_buildTupleStruct dts =+  let fnames = unsafeFieldName . pack. ("_" ++) . show <$> ([1..] :: [Int])+      fs = uncurry StructField <$> zip fnames dts+  in SQLType . StrictType . Struct . StructType $ V.fromList fs++-- instance (SQLTypeable a, SQLTypeable b) => SQLTypeable (a,b) where+--   _genericTypeFromValue _ = _genericTypeFromValue (undefined :: a) ++ _genericTypeFromValue (undefined :: b)++-- Generic SQLTypeable+class GenSQLTypeable a where+  genTypeFromProxy :: Proxy a -> GenericType++-- Datatype+instance GenSQLTypeable f => GenSQLTypeable (M1 D x f) where+  genTypeFromProxy _ = genTypeFromProxy (Proxy :: Proxy f)++-- Constructor Metadata+instance (GenSQLTypeable f, Constructor c) => GenSQLTypeable (M1 C c f) where+  genTypeFromProxy _+    | conIsRecord (undefined :: t c f a) =+        let !dt = genTypeFromProxy (Proxy :: Proxy f) in+          dt+    | otherwise =+        -- It is assumed to be a newtype and we are going to unwrap it+        let !dt1 = genTypeFromProxy (Proxy :: Proxy f)+        in case iSingleField (unSQLType dt1) of+          Just dt -> SQLType dt+          Nothing ->+            failure $ sformat ("M1 C "%sh%" dt1="%sh) n dt1+              where m = undefined :: t c f a+                    n = conName m++-- Selector Metadata+instance (GenSQLTypeable f, Selector c) => GenSQLTypeable (M1 S c f) where+  genTypeFromProxy _ =+    let !st = genTypeFromProxy (Proxy :: Proxy f)+        m = undefined :: t c f a+        n = selName m+        SQLType innerdt = st+        field = StructField { structFieldName = FieldName $ pack n, structFieldType = innerdt }+        st2 = StructType (V.singleton field) in+      SQLType (StrictType $ Struct st2)++-- Constructor Paramater+instance (GenSQLTypeable (Rep f), SQLTypeable f) => GenSQLTypeable (K1 R f) where+  genTypeFromProxy _ = _genericTypeFromValue (undefined :: f)++-- Sum branch+instance (GenSQLTypeable a, GenSQLTypeable b) => GenSQLTypeable (a :+: b) where+  genTypeFromProxy _ =+    let !y1 = genTypeFromProxy (Proxy :: Proxy a)+        !y2 = genTypeFromProxy (Proxy :: Proxy b) in+      -- TODO: need to prune the branch and throw an error here+      trace ("SUM: y1=" ++ show y1 ++ " y2=" ++ show y2) y1++-- Product branch+instance (GenSQLTypeable a, GenSQLTypeable b) => GenSQLTypeable (a :*: b) where+  genTypeFromProxy _ =+    let y1 = genTypeFromProxy (Proxy :: Proxy a)+        y2 = genTypeFromProxy (Proxy :: Proxy b) in case (y1, y2) of+        (SQLType (StrictType (Struct s1)), SQLType (StrictType (Struct s2))) ->+          (SQLType . StrictType . Struct) s where+            fs = structFields s1 V.++ structFields s2+            s = StructType fs+        _ -> failure $ sformat ("should not happen: left="%sh%" right="%sh) y1 y2++-- Void branch+instance GenSQLTypeable U1 where+  genTypeFromProxy _ = failure "U1"
+ src/Spark/Core/Internal/TypesStructures.hs view
@@ -0,0 +1,169 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}++{-| The structures of data types in Krapsh.++For a detailed description of the supported types, see+http://spark.apache.org/docs/latest/sql-programming-guide.html#data-types++At a high-level, Spark DataFrames and Datasets are equivalent to lists of+objects whose type can be mapped to the same StructType:+Dataset a ~ ArrayType StructType (...)+Columns of a dataset are equivalent to lists of object whose type can be+mapped to the same DataType (either Strict or Nullable)+Local data (or "blobs") are single elements whose type can be mapped to a+DataType (either strict or nullable)+-}+module Spark.Core.Internal.TypesStructures where++import Data.Aeson+import Data.Vector(Vector)+import qualified Data.Vector as V+import qualified Data.Aeson as A+import qualified Data.Text as T+import GHC.Generics(Generic)+import Test.QuickCheck++import Spark.Core.StructuresInternal(FieldName(..))+import Spark.Core.Internal.Utilities+++-- The core type algebra++-- | The data types that are guaranteed to not be null: evaluating them will return a value.+data StrictDataType =+    IntType+  | StringType+  | Struct !StructType+  | ArrayType { elementType :: !DataType } deriving (Eq)++-- | All the data types supported by the Spark engine.+-- The data types can either be nullable (they may contain null values) or strict (all the values are present).+-- There are a couple of differences with the algebraic data types in Haskell:+-- Maybe (Maybe a) ~ Maybe a which implies that arbitrary nesting of values will be flattened to a top-level Nullable+-- Similarly, [[]] ~ []+data DataType =+    StrictType !StrictDataType+  | NullableType !StrictDataType deriving (Eq)++-- | A field in a structure+data StructField = StructField {+  structFieldName :: !FieldName,+  structFieldType :: !DataType+} deriving (Eq)++-- | The main structure of a dataframe or a dataset+data StructType = StructType {+  structFields :: !(Vector StructField)+} deriving (Eq)+++-- Convenience types++-- | Represents the choice between a strict and a nullable field+data Nullable = CanNull | NoNull deriving (Show, Eq)++-- | Encodes the type of all the nullable data types+data NullableDataType = NullableDataType !StrictDataType deriving (Eq)++-- | A tagged datatype that encodes the sql types+-- This is the main type information that should be used by users.+data SQLType a = SQLType {+  -- | The underlying data type.+  unSQLType :: !DataType+} deriving (Eq, Generic)+++instance Show DataType where+  show (StrictType x) = show x+  show (NullableType x) = show x ++ "?"++instance Show StrictDataType where+  show StringType = "string"+  show IntType = "int"+  show (Struct struct) = show struct+  show array = "[" ++ show (elementType array) ++ "]"++instance Show StructField where+  show field = (T.unpack . unFieldName . structFieldName) field ++ ":" ++ s where+    s = show $ structFieldType field++instance Show StructType where+  show struct = "{" ++ unwords (map show (V.toList . structFields $ struct)) ++ "}"++instance Show (SQLType a) where+  show (SQLType dt) = show dt+++-- QUICKCHECK INSTANCES+++instance Arbitrary StructField where+  arbitrary = do+    name <- elements ["_1", "a", "b", "abc"]+    dt <- arbitrary :: Gen DataType+    return $ StructField (FieldName $ T.pack name) dt++instance Arbitrary StructType where+  arbitrary = do+    fields <- listOf arbitrary+    return . StructType . V.fromList $ fields++instance Arbitrary StrictDataType where+  arbitrary = do+    idx <- elements [1,2] :: Gen Int+    return $ case idx of+      1 -> StringType+      2 -> IntType+      _ -> failure "Arbitrary StrictDataType"++instance Arbitrary DataType where+  arbitrary = do+    x <- arbitrary+    u <- arbitrary+    return $ if x then+      StrictType u+    else+      NullableType u++-- AESON INSTANCES++-- This follows the same structure as the JSON generated by Spark.+instance ToJSON StrictDataType where+  toJSON IntType = "integer"+  toJSON StringType = "string"+  toJSON (Struct struct) = toJSON struct+  toJSON (ArrayType (StrictType dt)) =+    object [ "type" .= A.String "array"+           , "elementType" .= toJSON dt+           , "containsNull" .= A.Bool False ]+  toJSON (ArrayType (NullableType dt)) =+    object [ "type" .= A.String "array"+           , "elementType" .= toJSON dt+           , "containsNull" .= A.Bool True ]++instance ToJSON StructType where+  toJSON (StructType fields) =+    let+      fs = (snd . _fieldToJson) <$> V.toList fields+    in object [ "type" .= A.String "struct"+              , "fields" .= fs ]++-- Spark drops the info at the highest level.+instance ToJSON DataType where+  toJSON (StrictType dt) = object [+    "nullable" .= A.Bool False,+    "dt" .= toJSON dt]+  toJSON (NullableType dt) = object [+    "nullable" .= A.Bool True,+    "dt" .= toJSON dt]++_fieldToJson :: StructField -> (T.Text, A.Value)+_fieldToJson (StructField (FieldName n) (StrictType dt)) =+  (n, object [ "name" .= A.String n+             , "type" .= toJSON dt+             , "nullable" .= A.Bool False])+_fieldToJson (StructField (FieldName n) (NullableType dt)) =+  (n, object [ "name" .= A.String n+             , "type" .= toJSON dt+             , "nullable" .= A.Bool True])
+ src/Spark/Core/Internal/Utilities.hs view
@@ -0,0 +1,132 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE OverloadedStrings #-}+++{-| A collection of small utility functions.+-}+module Spark.Core.Internal.Utilities(+  LB.HasCallStack,+  UnknownType,+  pretty,+  myGroupBy,+  myGroupBy',+  CanRename(..),+  missing,+  failure,+  failure',+  forceRight,+  show',+  encodeDeterministicPretty,+  strictList,+  traceHint,+  SF.sh,+  (<&>),+  (<>)+  ) where++import Data.Aeson+import Data.Aeson.Encode.Pretty+import qualified Data.ByteString.Lazy.Char8 as Char8+import qualified Data.ByteString.Lazy as LBS+import qualified Data.Text as T+import qualified Formatting.ShortFormatters as SF+import Control.Arrow ((&&&))+import Data.List+import Data.Function+import Data.Text(Text)+import Formatting+import Debug.Trace(trace)+import qualified Data.Map.Strict as M+import Data.Monoid((<>))++import qualified Spark.Core.Internal.LocatedBase as LB++(<&>) :: Functor f => f a -> (a -> b) -> f b+(<&>) = flip fmap++-- | A type that is is not known and that is not meant to be exposed to the+-- user.+data UnknownType++{-| Pretty printing for Aeson values (and deterministic output)+-}+pretty :: Value -> Text+pretty = T.pack . Char8.unpack . encodeDeterministicPretty++-- | Produces a bytestring output of a JSON value that is deterministic+-- and that is invariant to the insertion order of the keys.+-- (i.e the keys are stored in alphabetic order)+-- This is to ensure that all id computations are stable and reproducible+-- on the server part.+-- TODO(kps) use everywhere JSON is converted+encodeDeterministicPretty :: Value -> LBS.ByteString+encodeDeterministicPretty =+  encodePretty' (defConfig { confIndent = Spaces 0, confCompare = compare })++-- | group by+myGroupBy' :: (Ord b) => (a -> b) -> [a] -> [(b, [a])]+myGroupBy' f = map (f . head &&& id)+                   . groupBy ((==) `on` f)+                   . sortBy (compare `on` f)++-- | group by+myGroupBy :: (Ord a) => [(a, b)] -> M.Map a [b]+myGroupBy l = let+  l2 = myGroupBy' fst l in+  M.map (snd <$>) $ M.fromList l2++-- | The class of types that can be renamed.+-- It is made generic because it covers 2 notions:+--   - the name of a compute node that will eventually determine its compute path+--   - the name of field (which may become an object path)+-- This syntax tries to be convenient and will fail immediately+-- for basic errors such as illegal characters.+--+-- This could be revisited in the future, but it is a compromise+-- on readability.+class CanRename a txt where+  (@@) :: a -> txt -> a++infixl 1 @@++-- | Missing implementations in the code base.+missing :: (LB.HasCallStack) => Text -> a+missing msg = LB.error $ T.concat ["MISSING IMPLEMENTATION: ", msg]++{-| The function that is used to trigger exception due to internal programming+errors.++Currently, all programming errors simply trigger an exception. All these+impure functions are tagged with an implicit call stack argument.+-}+failure :: (LB.HasCallStack) => Text -> a+failure msg = LB.error (T.concat ["FAILURE in Spark. Hint: ", msg])++failure' :: (LB.HasCallStack) => Format Text (a -> Text) -> a -> c+failure' x = failure . sformat x+++{-| Given a DataFrame or a LocalFrame, attempts to get the value,+or throws the error.++This function is not total.+-}+forceRight :: (LB.HasCallStack, Show a) => Either a b -> b+forceRight (Right b) = b+forceRight (Left a) = LB.error $+  sformat ("Failure from either, got instead a left: "%shown) a++-- | Force the complete evaluation of a list to WNF.+strictList :: (Show a) => [a] -> [a]+strictList [] = []+strictList (h : t) = let !t' = strictList t in (h : t')++-- | (internal) prints a hint with a value+traceHint :: (Show a) => Text -> a -> a+traceHint hint x = trace (T.unpack hint ++ show x) x++-- | show with Text+show' :: (Show a) => a -> Text+show' x = T.pack (show x)
+ src/Spark/Core/Row.hs view
@@ -0,0 +1,13 @@+module Spark.Core.Row(+  module Spark.Core.Internal.RowStructures,+  ToSQL,+  FromSQL,+  valueToCell,+  cellToValue,+  jsonToCell+  ) where++import Spark.Core.Internal.RowStructures+import Spark.Core.Internal.RowGenerics+import Spark.Core.Internal.RowGenericsFrom+import Spark.Core.Internal.RowUtils
+ src/Spark/Core/StructuresInternal.hs view
@@ -0,0 +1,109 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}++-- Some basic structures about nodes in a graph, etc.++module Spark.Core.StructuresInternal(+  NodeName(..),+  NodePath(..),+  NodeId(..),+  FieldName(..),+  FieldPath(..),+  catNodePath,+  fieldName,+  unsafeFieldName,+  fieldPath,+) where++import qualified Data.Text as T+import Data.ByteString(ByteString)+import GHC.Generics (Generic)+import Data.Hashable(Hashable)+import Data.List(intercalate)+import qualified Data.Aeson as A+import Data.String(IsString(..))+import Data.Vector(Vector)+import qualified Data.Vector as V++import Spark.Core.Internal.Utilities++-- | The name of a node (without path information)+newtype NodeName = NodeName { unNodeName :: T.Text } deriving (Eq, Ord)++-- | The user-defined path of the node in the hierarchical representation of the graph.+newtype NodePath = NodePath { unNodePath :: Vector NodeName } deriving (Eq)++-- | The unique ID of a node. It is based on the parents of the node+-- and all the relevant intrinsic values of the node.+newtype NodeId = NodeId { unNodeId :: ByteString } deriving (Eq, Ord, Generic)++-- | The name of a field in a sql structure+-- This structure ensures that proper escaping happens if required.+-- TODO: prevent the constructor from being used, it should be checked first.+newtype FieldName = FieldName { unFieldName :: T.Text } deriving (Eq)++-- | A path to a nested field an a sql structure.+-- This structure ensures that proper escaping happens if required.+newtype FieldPath = FieldPath { unFieldPath :: Vector FieldName } deriving (Eq)++-- | A safe constructor for field names that fixes all the issues relevant to+-- SQL escaping+-- TODO: proper implementation+fieldName :: T.Text -> Either String FieldName+fieldName = Right . FieldName++-- | Constructs the field name, but will fail if the content is not correct.+unsafeFieldName :: (HasCallStack) => T.Text -> FieldName+unsafeFieldName = forceRight . fieldName++-- | A safe constructor for field names that fixes all the issues relevant to SQL escaping+-- TODO: proper implementation+fieldPath :: T.Text -> Either String FieldPath+fieldPath x = Right . FieldPath . V.singleton $ FieldName x++-- | The concatenated path. This is the inverse function of fieldPath.+catNodePath :: NodePath -> T.Text+catNodePath (NodePath np) =+  T.intercalate "/" (unNodeName <$> V.toList np)++instance Show NodeId where+  show (NodeId bs) = let s = show bs in+    if length s > 9 then+      (drop 1 . take 6) s ++ ".."+    else+      s++instance Show NodeName where+  show (NodeName nn) = T.unpack nn++instance Show NodePath where+  show np = T.unpack $ T.concat ["NPath(", catNodePath np, ")" ]++instance Show FieldPath where+  show (FieldPath l) =+    intercalate "." (show <$> V.toList l)++instance Show FieldName where+  -- TODO(kps) escape the '.' characters in the field name+  show (FieldName fn) = T.unpack fn++instance Hashable NodeId++instance IsString FieldName where+  fromString = FieldName . T.pack++instance A.ToJSON NodeName where+  toJSON = A.toJSON . unNodeName++instance A.ToJSON NodePath where+  toJSON = A.toJSON . unNodePath++instance A.ToJSON FieldName where+  toJSON = A.toJSON . unFieldName++instance A.ToJSON FieldPath where+  toJSON = A.toJSON . unFieldPath++instance Ord FieldName where+  compare f1 f2 = compare (unFieldName f1) (unFieldName f2)
+ src/Spark/Core/Try.hs view
@@ -0,0 +1,51 @@++{-|+Useful classes and functions to deal with failures+within the Krapsh framework.++This is a developer API. Users should not have to invoke functions+from this module.+-}+module Spark.Core.Try(+  NodeError(..),+  Try,+  tryError,+  tryEither+  ) where++import qualified Data.Text as T+import qualified Data.Vector as V++import Spark.Core.StructuresInternal++-- | An error associated to a particular node (an observable or a dataset).+data NodeError = Error {+  ePath :: NodePath,+  eMessage :: T.Text+} deriving (Eq, Show)++-- | The common result of attempting to build something.+type Try = Either NodeError+++-- TODO: rename to tryError+_error :: T.Text -> Try a+_error txt = Left Error {+    ePath = NodePath V.empty,+    eMessage = txt+  }++-- | Returns an error object given a text clue.+tryError :: T.Text -> Try a+tryError = _error++-- | Returns an error object given a string clue.+--Remove this method+--tryError' :: String -> Try a+--tryError' = _error . T.pack++-- | (internal)+-- Given a potentially errored object, converts it to a Try.+tryEither :: Either T.Text a -> Try a+tryEither (Left msg) = tryError msg+tryEither (Right x) = Right x
+ src/Spark/Core/Types.hs view
@@ -0,0 +1,53 @@+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Spark.Core.Types(+  DataType,+  Nullable(..),+  TupleEquivalence(..),+  NameTuple(..),+  -- intType,+  -- canNull,+  -- noNull,+  -- stringType,+  -- arrayType',+  -- cellType,+  -- structField,+  -- structType,+  -- arrayType,+  SQLType,+  columnType,+  SQLTypeable,+  buildType,+  StructField,+  StructType,+  castType,+  catNodePath+) where++import Formatting++import Spark.Core.Internal.TypesStructures+import Spark.Core.Internal.TypesGenerics+import Spark.Core.Internal.TypesFunctions+import Spark.Core.StructuresInternal+import Spark.Core.Internal.FunctionsInternals(TupleEquivalence(..), NameTuple(..))+import Spark.Core.Try++-- | Description of types supported in DataSets+-- Krapsh supports a restrictive subset of Algebraic Datatypes that is amenable to SQL+-- transformations. This file contains the description of all the supported types, and some+-- conversion tools.+++-- -- Converts an (untyped) datatype to a generic tagged SQLType.+-- cellType :: DataType -> CellType+-- cellType = SQLType++-- Takes a given type and attempts to cast it to another type,+-- which is known by the type system.+castType :: forall a b. (SQLTypeable b) => SQLType a -> Try (SQLType b)+castType sqlt =+  let sqlt' = buildType :: SQLType b in+    if unSQLType sqlt == unSQLType sqlt' then Right sqlt'+      else tryError $ sformat ("castType: tried to cast "%shown%" into incompatible type "%shown) sqlt sqlt'
+ test-integration/Spark/Core/CachingSpec.hs view
@@ -0,0 +1,41 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.CachingSpec where++import Test.Hspec+import qualified Data.Text++import Spark.Core.Context+import Spark.Core.Functions+import Spark.Core.Column+import Spark.Core.ColumnFunctions+++-- Collecting a dataset made from a list should yield the same list (modulo+-- some reordering)+collectIdempotent :: [Int] -> IO ()+collectIdempotent l = do+  let ds = dataset l+  let ds' = autocache ds+  let c1 = asCol ds'+  let s1 = colSum c1+  let s2 = count ds'+  let x = s1 + s2+  l2 <- exec1Def x+  l2 `shouldBe` (sum l + length l)++run :: String -> IO () -> SpecWith (Arg (IO ()))+run s f = it s $ do+  createSparkSessionDef $ defaultConf { confRequestedSessionName = Data.Text.pack s }+  f+  -- This is horribly not robust to any sort of failure, but it will do for now+  -- TODO(kps) make more robust+  closeSparkSessionDef+  return ()++spec :: Spec+spec = do+  describe "Integration test - caching on ints" $ do+    run "cache_sum1" $+      collectIdempotent ([1,2,3] :: [Int])
+ test-integration/Spark/Core/CollectSpec.hs view
@@ -0,0 +1,56 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.CollectSpec where++import Test.Hspec+import qualified Data.Text+import Data.List(sort)++import Spark.Core.Context+import Spark.Core.Types+import Spark.Core.Row+import Spark.Core.Functions+import Spark.Core.Column+++-- Collecting a dataset made from a list should yield the same list (modulo+-- some reordering)+-- TODO: replace the ordering by the canonical ordering over the data+collectIdempotent :: (Ord a, Eq a, Show a, SQLTypeable a, ToSQL a, FromSQL a) => [a] -> IO ()+collectIdempotent l = do+  let ds = dataset l+  l2 <- exec1Def $ collect (asCol ds)+  l2 `shouldBe` sort l++run :: String -> IO () -> SpecWith (Arg (IO ()))+run s f = it s $ do+  createSparkSessionDef $ defaultConf { confRequestedSessionName = Data.Text.pack s }+  f+  -- This is horribly not robust to any sort of failure, but it will do for now+  -- TODO(kps) make more robust+  closeSparkSessionDef+  return ()++spec :: Spec+spec = do+  describe "Integration test - collect on ints" $ do+    run "empty_ints1" $+      collectIdempotent ([] :: [Int])+    run "ints1" $+      collectIdempotent ([4,5,1,2,3] :: [Int])+    -- TODO(kps) in Spark 2.0.2, this fails!!!+    -- Works with Spark 2.0.1 -> report+    -- run "ints1_opt" $+    --   collectIdempotent ([Just 1, Nothing] :: [Maybe Int])+    -- run "nothing_ints_opt" $+    --   collectIdempotent ([Nothing] :: [Maybe Int])+    run "ints1_opt" $+      collectIdempotent ([Just 1, Just 2] :: [Maybe Int])+    run "empty_ints_opt" $+      collectIdempotent ([] :: [Maybe Int])+  -- describe "Integration test - collect on TestStruct5" $ do+  --   run "empty_TestStruct5" $+  --     collectIdempotent ([] :: [TestStruct5])+  --   run "empty_TestStruct5" $+  --     collectIdempotent ([TestStruct5 1 2] :: [TestStruct5])
+ test-integration/Spark/Core/IntegrationUtilities.hs view
@@ -0,0 +1,53 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++module Spark.Core.IntegrationUtilities where++import GHC.Generics (Generic)++import Spark.Core.Context+import Spark.Core.Types+import Spark.Core.Row++data TestStruct1 = TestStruct1 {+  ts1f1 :: Int,+  ts1f2 :: Maybe Int } deriving (Show, Eq, Generic)+instance ToSQL TestStruct1+instance FromSQL TestStruct1+instance SQLTypeable TestStruct1++data TestStruct2 = TestStruct2 {+  ts2f1 :: [Int]+} deriving (Show, Generic)+instance SQLTypeable TestStruct2++data TestStruct3 = TestStruct3 {+  ts3f1 :: Int+} deriving (Show, Eq, Generic)+instance ToSQL TestStruct3+instance SQLTypeable TestStruct3++data TestStruct4 = TestStruct4 {+  ts4f1 :: TestStruct3+} deriving (Show, Eq, Generic)++data TestStruct5 = TestStruct5 {+  ts5f1 :: Int,+  ts5f2 :: Int+} deriving (Show, Eq, Generic)+-- instance ToJSON TestStruct5+instance SQLTypeable TestStruct5+instance FromSQL TestStruct5+instance ToSQL TestStruct5++data TestStruct6 = TestStruct6 {+  ts6f1 :: Int,+  ts6f2 :: Int,+  ts6f3 :: TestStruct3+} deriving (Show, Eq, Generic)++newtype TestT1 = TestT1 {+  unTestT1 :: Int+} deriving (Eq, Show, Generic, Num)
+ test-integration/Spark/Core/SimpleAddSpec.hs view
@@ -0,0 +1,67 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.SimpleAddSpec where++import Test.Hspec+import qualified Data.Text++import Spark.Core.Context+import Spark.Core.Types+import Spark.Core.Row+import Spark.Core.Functions+++-- Collecting a dataset made from a list should yield the same list (modulo+-- some reordering)+smallSum :: (Eq a, Show a, SQLTypeable a, ToSQL a, FromSQL a, Num a) => a -> a -> IO ()+smallSum x y = do+  let x' = constant x+  let y' = constant y+  let z1' = x' + y'+  z1 <- exec1Def z1'+  z1 `shouldBe` (x + y)+  let z2' = y' + x'+  z2 <- exec1Def z2'+  z2 `shouldBe` (x + y)++negation :: (Eq a, Show a, SQLTypeable a, ToSQL a, FromSQL a, Num a) => a -> a -> IO ()+negation x y = do+  let x' = constant x+  let y' = constant y+  let z1' = x' - y'+  z1 <- exec1Def z1'+  z1 `shouldBe` (x - y)+  let z2' = y' - x'+  z2 <- exec1Def z2'+  z2 `shouldBe` (y - x)++checkNegate :: (Eq a, Show a, SQLTypeable a, ToSQL a, FromSQL a, Num a) => a -> IO ()+checkNegate x = do+  let x' = constant x+  let z1' = negate x'+  z1 <- exec1Def z1'+  z1 `shouldBe` negate x++run :: String -> IO () -> SpecWith (Arg (IO ()))+run s f = it s $ do+  createSparkSessionDef $ defaultConf {+    confRequestedSessionName = Data.Text.pack s,+    confPollingIntervalMillis = 100 }+  f+  -- This is horribly not robust to any sort of failure, but it will do for now+  -- TODO(kps) make more robust+  closeSparkSessionDef+  return ()++spec :: Spec+spec = do+  describe "Integration test - sum on ints" $ do+    run "empty_ints1" $+      smallSum (1 :: Int) (2 :: Int)+    run "zero_ints1" $+      smallSum (0 :: Int) (2 :: Int)+    run "negation_ints1" $+      negation (1 :: Int) (2 :: Int)+    run "negate_ints1" $+      checkNegate (1 :: Int)
+ test-integration/Spec.hs view
@@ -0,0 +1,2 @@+-- Not working???+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}
+ test/Spark/Core/ColumnSpec.hs view
@@ -0,0 +1,37 @@++module Spark.Core.ColumnSpec where++import Test.Hspec++import Spark.Core.Column+import Spark.Core.Dataset+import Spark.Core.Functions+import Spark.Core.ColumnFunctions++data Z+data Y++spec :: Spec+spec = do+  describe "ColumnSpec: ensure rules compile correctly" $ do+    let c1 = undefined :: Column Z Int+    let c2 = undefined :: Column Z Int+    let c1' = undefined :: DynColumn+    let c2' = undefined :: DynColumn+    let i1 = 3 :: Int+    let o1 = constant 4 :: LocalData Int+    let o2 = 5 :: LocalData Int+    let o1' = undefined :: LocalFrame+    let o2' = undefined :: LocalFrame+    it "+ should not blow up" $ do+      let z1 = c1 + c2+      let z2 = c1' + c2'+      let z3 = c1 + 1+      let z4 = 1 + c1+      'a' `shouldBe` 'a'+    it ".+ should not blow up" $ do+      let z1 = c1 .+ c2+      let z2 = c1' .+ c2'+      let z3 = c1 .+ c2'+      let z4 = c1' .+ c2+      'a' `shouldBe` 'a'
+ test/Spark/Core/ContextSpec.hs view
@@ -0,0 +1,17 @@+++module Spark.Core.ContextSpec where++import Test.Hspec++import Spark.Core.Functions++spec :: Spec+spec = do+  describe "Basic routines to get something out" $ do+    it "should print a node" $ do+      let x = dataset ([1 ,2, 3, 4]::[Int])+      x `shouldBe` x+        --   b = nodeToBundle (untyped x) in+        -- trace (pretty b) $+        --   1 `shouldBe` 1
+ test/Spark/Core/DatasetSpec.hs view
@@ -0,0 +1,72 @@+-- {-# LANGUAGE OverloadedStrings #-}+-- {-# LANGUAGE ExtendedDefaultRules #-}+++module Spark.Core.DatasetSpec where++import qualified Data.Text as T+import Test.Hspec+import qualified Data.Vector as V++import Spark.Core.Dataset+import Spark.Core.Functions+import Spark.Core.Column+import Spark.Core.StructuresInternal+import Spark.Core.Internal.ContextInternal+import Spark.Core.Internal.Utilities+import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.ComputeDag++nName :: String -> NodeName+nName = NodeName . T.pack++spec :: Spec+spec = do+  describe "create a dataframe" $ do+    it "should not explode" $+      let x = dataset ([1 ,2, 3, 4]::[Int]) in+        nodeName x `shouldBe` nName "constant_1b99f8"++    it "renaming should work" $+      let x = dataset ([1 ,2, 3]::[Int]) @@ "ds1" in+        nodeName x `shouldBe` nName "ds1"++  describe "check localset" $ do+    it "should not explode" $+      let x = dataset ([1 ,2, 3]::[Int]) in+        nodeName x `shouldBe` nName "constant_e5125b"++    it "renaming should work" $+      let x = dataset ([1 ,2, 3]::[Int]) @@ "ds1" in+        nodeName x `shouldBe` nName "ds1"++  describe "column syntax" $+    it "should not explode" $ do+      let ds = dataset ([1 ,2, 3]::[Int])+      let c1 = ds // "c1"+      c1 `shouldBe` c1++  describe "Logical dependencies" $ do+    it "should work" $ do+      let ds = dataset ([1 ,2, 3, 4]::[Int])+      let ds1 = dataset ([1]::[Int]) `depends` [untyped ds]+      let g = traceHint (T.pack "g=") $ computeGraphToGraph $ forceRight $ buildComputationGraph ds1+      V.length (gVertices g) `shouldBe` 2+++--   describe "simple test" $ do+--     it "the type should match" $ do+--       let+--        n1 = constant "xxx"+--        n = NodeType $ T.pack "org.spark.Constant" in+--         (nodeOp n1) `shouldBe` n++--     it "no name" $ do+--       let n1 = constant "xxx"+--           t = NodeName $ T.pack "org.spark.Constant" in+--         (nodeName n1) `shouldBe` t++--     it "some name" $ do+--       let n1 = constant "xxx" @@ "name"+--           t = NodeName $ T.pack "name" in+--         (nodeName n1) `shouldBe` t
+ test/Spark/Core/Internal/CachingSpec.hs view
@@ -0,0 +1,164 @@+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.Internal.CachingSpec where++import Test.Hspec++import qualified Data.ByteString.Char8 as C8+import Data.Either(isLeft, isRight)+import Control.Arrow((&&&))+import Data.Text(Text)+import Data.Foldable(toList)+import Formatting++import Spark.Core.Try+import Spark.Core.Functions+import Spark.Core.Column+import Spark.Core.ColumnFunctions+import Spark.Core.Internal.Caching+-- Required for instance resolution+import Spark.Core.StructuresInternal()+import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.DAGFunctions+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.Utilities+import Spark.Core.Internal.ContextStructures+import Spark.Core.Internal.ContextInternal++data TestType = AutocacheNode | CacheNode | UncacheNode | Dataset | Row deriving (Eq, Show)++data TestNode = TestNode { tnId :: VertexId, tnType :: TestType, tnParents :: [(StructureEdge, TestNode)] } deriving (Eq)++instance Show TestNode where+  show v = "TestNode(" ++ (C8.unpack . unVertexId. tnId $ v) ++ ")"++nid :: String -> VertexId+nid = VertexId . C8.pack++node :: String -> TestType -> [TestNode] -> TestNode+node s tp p = TestNode (VertexId (C8.pack s)) tp ((const ParentEdge &&& id) <$> p)++instance GraphVertexOperations TestNode where+  vertexToId = tnId+  expandVertexAsVertices x = snd <$> tnParents x++instance GraphOperations TestNode StructureEdge where+  expandVertex tn = tnParents tn++acGen :: AutocacheGen TestNode+acGen =+  let deriveUncache' (Vertex (VertexId x) (TestNode _ AutocacheNode _)) =+        let vid' = VertexId $ C8.pack . (++ "_uncache") . C8.unpack $ x+        in Vertex vid' (TestNode vid' UncacheNode [])+      deriveUncache' x = error (show x)+      deriveIdentity' (Vertex (VertexId x) (TestNode _ r _)) =+        let vid' = VertexId $ C8.pack . (++ "_identity") . C8.unpack $ x+        in Vertex vid' (TestNode vid' r [])+  in AutocacheGen {+    deriveUncache = deriveUncache',+    deriveIdentity = deriveIdentity'+  }++expandFun :: TestNode -> CacheTry NodeCachingType+expandFun n = case (tnType n, tnParents n) of+  (AutocacheNode, [_]) -> pure $ AutocacheOp (tnId n)+  (AutocacheNode, x) -> Left $ sformat ("Node: "%shown%": expected one parent for autocaching, got "%shown) n x+  (CacheNode, [_]) -> pure $ CacheOp (tnId n)+  (CacheNode, x) -> Left $ sformat ("Node: "%shown%": expected one parent for caching, got "%shown) n x+  (UncacheNode, [(ParentEdge, x)]) -> pure $ UncacheOp (tnId n) (tnId x)+  (UncacheNode, x) -> Left $ sformat ("Node: "%shown%": Expected one parent for uncaching, got "%shown) n x+  (Dataset, _) -> Right Through+  (Row, _) -> Right Stop++errors :: TestNode -> CacheTry [CachingFailure]+errors tn = do+  g <- buildGraph tn :: Either Text (Graph TestNode StructureEdge)+  checkCaching (graphMapEdges g (const ParentEdge)) expandFun++errors' :: TestNode -> CacheTry (Graph TestNode StructureEdge)+errors' tn = do+  g <- buildGraph tn :: Either Text (Graph TestNode StructureEdge)+  fillAutoCache expandFun acGen g++intErrors :: LocalData a -> Try ComputeGraph+intErrors ld =+  let cg = buildComputationGraph ld+  in performGraphTransforms =<< cg+++spec :: Spec+spec = do+  describe "Caching operations" $ do+    it "missing parent node" $ do+      let n1 = node "1" CacheNode []+      errors n1 `shouldSatisfy` isLeft+    it "caching: parent node" $ do+      let n0 = node "0" Dataset []+      let n1 = node "1" CacheNode [n0]+      errors n1 `shouldBe` Right []+    it "uncaching: missing parent node" $ do+      let n1 = node "1" UncacheNode []+      errors n1 `shouldSatisfy` isLeft+    it "uncaching: parent node" $ do+      let n0 = node "0" Dataset []+      let n1 = node "1" CacheNode [n0]+      let n2 = node "2" CacheNode [n1]+      errors n2 `shouldBe` Right []+    it "too many nodes for uncaching" $ do+      let n0 = node "0" Dataset []+      let n1 = node "1" CacheNode [n0]+      let n2 = node "2" UncacheNode [n1, n2]+      errors n2 `shouldSatisfy` isLeft+    it "access after uncaching" $ do+      let n0 = node "0" Dataset []+      let n1 = node "1" CacheNode [n0]+      let n2 = node "2" UncacheNode [n1]+      let n3 = node "3" Dataset [n1, n2]+      errors n3 `shouldBe` Right [CachingFailure (nid "1") (nid "2") (nid "3")]+    it "ambigous access after uncaching" $ do+      let n0 = node "0" Dataset []+      let n1 = node "1" CacheNode [n0]+      let n2 = node "2" UncacheNode [n1]+      let n3 = node "3" Dataset [n1]+      let n4 = node "4" Dataset [n3, n2]+      errors n4 `shouldBe` Right [CachingFailure (nid "1") (nid "2") (nid "3")+                                  ,CachingFailure (nid "1") (nid "2") (nid "4")]+  describe "Autocaching operations" $ do+    it "missing parent node" $ do+      let n1 = node "1" AutocacheNode []+      let g = traceHint "g=" (errors' n1)+      g `shouldSatisfy` isLeft+    it "auto-uncaching with no child should not create uncaching" $ do+      let n0 = node "0" Dataset []+      let n1 = node "1" AutocacheNode [n0]+      let g = traceHint "g=" (errors' n1)+      g `shouldSatisfy` isRight+      ((length . toList . gVertices) <$> g) `shouldBe` Right 2+    it "access after uncaching" $ do+      let n0 = node "0" Dataset []+      let n1 = node "1" AutocacheNode [n0]+      let n2 = node "2" Row [n1]+      let g = traceHint "g=" (errors' n2)+      g `shouldSatisfy` isRight+      ((length . toList . gVertices) <$> g) `shouldBe` Right 5+    it "access after and scoping" $ do+      let n0 = node "0" Dataset []+      let n1 = node "1" AutocacheNode [n0]+      let n2a = node "2a" Row [n1]+      let n2b = node "2b" Row [n1]+      let n3 = node "3" Row [n2a, n2b]+      let g = traceHint "g=" (errors' n3)+      g `shouldSatisfy` isRight+      ((length . toList . gVertices) <$> g) `shouldBe` Right 8+  describe "Autocaching integration tests" $ do+    it "test 1" $ do+      let l = [1,2,3] :: [Int]+      let ds = dataset l+      let ds' = autocache ds+      let c1 = asCol ds'+      let s1 = colSum c1+      let s2 = count ds'+      let x = s1 + s2+      let g = traceHint "g=" (intErrors x)+      g `shouldSatisfy` isRight
+ test/Spark/Core/Internal/DAGFunctionsSpec.hs view
@@ -0,0 +1,113 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE MultiParamTypeClasses #-}++-- :script test/Spark/Core/Internal/PathsSpec.hs+module Spark.Core.Internal.DAGFunctionsSpec where++import Test.Hspec+import qualified Data.Map.Strict as M+import qualified Data.Vector as V+import qualified Data.ByteString.Char8 as C8+import Control.Arrow((&&&))+import Data.Foldable(toList)++import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.DAGFunctions+import Spark.Core.Internal.Utilities++data MyV = MyV {+  mvId :: VertexId,+  mvParents :: [MyV]+} deriving (Eq)++id2Str :: VertexId -> String+id2Str = C8.unpack . unVertexId++instance Show MyV where+  show v = "MyV(" ++ (id2Str . mvId $ v) ++ ")"+++instance GraphVertexOperations MyV where+  vertexToId = mvId+  expandVertexAsVertices = mvParents++instance GraphOperations MyV () where+  expandVertex = ((const () &&& id) <$>) . mvParents++myv :: String -> [MyV] -> MyV+myv s = MyV (VertexId (C8.pack s))++expandNodes :: MyV -> DagTry [String]+expandNodes vx =+  let tg = buildGraph vx :: DagTry (Graph MyV ())+  in (id2Str . mvId . vertexData <$>) . toList . gVertices <$> tg++-- edges: from -> to+expandEdges :: MyV -> DagTry [(String, String)]+expandEdges vx =+  let tg = buildGraph vx :: DagTry (Graph MyV ())+  in tg <&> \g ->+    concat $ M.assocs (gEdges g) <&> \(vid, v) ->+      (C8.unpack . unVertexId . vertexId . veEndVertex &&&+       C8.unpack . unVertexId . const vid) <$> V.toList v++spec :: Spec+spec = do+  describe "Tests on paths" $ do+    it "no parent" $ do+      let v0 = myv "v0" []+      expandNodes v0 `shouldBe` Right ["v0"]+    it "common parent" $ do+      let v0 = myv "v0" []+      let v0' = myv "v0" []+      let v1 = myv "v1" [v0, v0']+      expandEdges v1 `shouldBe` Right [("v0", "v1"), ("v0", "v1")]+    it "diamond" $ do+      let va = myv "va" []+      let va' = myv "va" []+      let v0 = myv "v0" [va]+      let v0' = myv "v0" [va']+      let v1 = myv "v1" [v0, v0']+      expandEdges v1 `shouldBe` Right [("va", "v0"), ("v0", "v1"), ("v0", "v1")]+    it "simple sources" $ do+      let v0 = myv "v0" []+      let v1 = myv "v1" [v0]+      let tg = buildGraph v1 :: DagTry (Graph MyV ())+      let g = forceRight tg+      mvId . vertexData <$> graphSources g `shouldBe` [mvId v1]+    it "simple sinks" $ do+      let v0 = myv "v0" []+      let v1 = myv "v1" [v0]+      let tg = buildGraph v1 :: DagTry (Graph MyV ())+      let g = forceRight tg+      mvId . vertexData <$> graphSinks g `shouldBe` [mvId v0]+    it "longer sources" $ do+      let v0 = myv "v0" []+      let v1 = myv "v1" [v0]+      let v2 = myv "v2" [v1]+      let tg = buildGraph v2 :: DagTry (Graph MyV ())+      let g = forceRight tg+      mvId . vertexData <$> graphSources g `shouldBe` [mvId v2]+    it "longer sinks" $ do+      let v0 = myv "v0" []+      let v1 = myv "v1" [v0]+      let v2 = myv "v2" [v1]+      let tg = buildGraph v2 :: DagTry (Graph MyV ())+      let g = forceRight tg+      mvId . vertexData <$> graphSinks g `shouldBe` [mvId v0]+  describe "building DAGs" $ do+    it "2 nodes" $ do+      let v0 = myv "v0" []+      let v1 = myv "v1" [v0]+      let v2 = myv "v2" [v1]+      let l = forceRight $ buildVertexList v2+      id2Str . mvId <$> l `shouldBe` ["v0", "v1", "v2"]+    it "triangle" $ do+      let v0 = myv "v0" []+      let v1 = myv "v1" [v0]+      let v2 = myv "v2" [v0, v1]+      let l = forceRight $ buildVertexList v2+      -- The return order should be in lexicographic order+      -- (which is unique in this case).+      id2Str . mvId <$> l `shouldBe` ["v0", "v1", "v2"]
+ test/Spark/Core/Internal/LocalDataFunctionsSpec.hs view
@@ -0,0 +1,18 @@++module Spark.Core.Internal.LocalDataFunctionsSpec where++import Test.Hspec++import Spark.Core.Dataset+import Spark.Core.Functions()++spec :: Spec+spec = do+  describe "Arithmetic operations on local data (integers)" $ do+    it "ints" $ do+      let x1 = 1 :: LocalData Int+      let x2 = 2 :: LocalData Int+      let y1 = x1 + x2+      let y2 = x1 `div` x2+      (y2 `shouldBe` y2)+      (y1 `shouldBe` y1)
+ test/Spark/Core/Internal/OpFunctionsSpec.hs view
@@ -0,0 +1,33 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE QuasiQuotes #-}++module Spark.Core.Internal.OpFunctionsSpec where++import Data.Aeson+import Test.Hspec+import Text.RawString.QQ++import Spark.Core.Functions+import Spark.Core.Internal.OpFunctions+import Spark.Core.Internal.DatasetFunctions+++spec :: Spec+spec = do+  describe "extraNodeOpData" $ do+    it "should have the content of a constant dataset" $ do+      let l = [1,2,3] :: [Int]+      let res :: Maybe Value+          res = decode+              ([r|{"content": [1,2,3],+                    "cellType" : {+                      "dt": "integer",+                      "nullable": false+                    }+                  }|])+      let ds = dataset l+      let d = extraNodeOpData . nodeOp $ ds+      Just d `shouldBe` res
+ test/Spark/Core/Internal/PathsSpec.hs view
@@ -0,0 +1,188 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- :script test/Spark/Core/Internal/PathsSpec.hs+module Spark.Core.Internal.PathsSpec where++import Test.Hspec+import qualified Data.Map.Strict as M+import qualified Data.Set as S+import qualified Data.ByteString.Char8 as C8+import qualified Data.Text as T++import Spark.Core.StructuresInternal+import Spark.Core.Functions+import Spark.Core.Dataset+import Spark.Core.Internal.Paths+import Spark.Core.Internal.DAGStructures+import Spark.Core.Internal.DAGFunctions+import Spark.Core.Internal.ComputeDag+import Spark.Core.Internal.PathsUntyped+import Spark.Core.Internal.Utilities+import Spark.Core.Internal.DatasetFunctions+import Spark.Core.Internal.DatasetStructures++data MyV = MyV {+  mvId :: VertexId,+  mvLogical :: [MyV],+  mvParents :: [MyV]+} deriving (Eq)++instance Show MyV where+  show v = "MyV(" ++ (C8.unpack . unVertexId . mvId $ v) ++ ")"+++assignPaths :: UntypedNode -> [UntypedNode]+assignPaths n =+  let cgt = buildCGraph n :: DagTry (ComputeDag UntypedNode NodeEdge)+      cg = forceRight cgt+      acgt = assignPathsUntyped cg+      ncg = forceRight acgt+  in graphDataLexico . tieNodes $ ncg+++instance GraphVertexOperations MyV where+  vertexToId = mvId+  expandVertexAsVertices = mvParents++myv :: String -> [MyV] -> [MyV] -> MyV+myv s logical inner = MyV (VertexId (C8.pack s)) logical inner++myvToVertex :: MyV -> Vertex MyV+myvToVertex x = Vertex (mvId x) x++buildScopes :: [MyV] -> Scopes+buildScopes l = iGetScopes0 l' fun where+  l' = myvToVertex <$> l+  fun vx = ParentSplit {+    psLogical = myvToVertex <$> (mvLogical . vertexData $ vx),+    psInner = myvToVertex <$> (mvParents . vertexData $ vx) }++simple :: [(Maybe String, [String])] -> Scopes+simple [] = M.empty+simple ((ms, ss) : t) =+  let+    key = VertexId . C8.pack <$> ms+    vals = VertexId . C8.pack <$> ss+    new = M.singleton key (S.fromList vals)+  in mergeScopes new (simple t)++gatherings :: [(String, [[String]])] -> M.Map VertexId [[VertexId]]+gatherings [] = M.empty+gatherings ((key, paths) : t) =+  let+    k = VertexId . C8.pack $ key+    ps = (VertexId . C8.pack <$>) <$> paths+    new = M.singleton k ps+  in M.unionWith (++) new (gatherings t)++gatherPaths' :: [MyV] -> M.Map VertexId [[VertexId]]+gatherPaths' = gatherPaths . buildScopes++spec :: Spec+spec = do+  describe "Tests on paths" $ do+    it "nothing" $ do+      buildScopes [] `shouldBe` simple []+    it "no parent" $ do+      let v0 = myv "v0" [] []+      let res = [ (Nothing, ["v0"]), (Just "v0", []) ]+      buildScopes [v0] `shouldBe` simple res+    it "one logical parent" $ do+      let v0 = myv "v0" [] []+      let v1 = myv "v1" [v0] []+      let res = [ (Nothing, ["v0", "v1"])+                , (Just "v1", [])+                , (Just "v0", []) ]+      buildScopes [v1, v0] `shouldBe` simple res+    it "one inner parent" $ do+      let v0 = myv "v0" [] []+      let v1 = myv "v1" [] [v0]+      let res = [ (Nothing, ["v1"])+                , (Just "v1", ["v0"])+                , (Just "v0", []) ]+      buildScopes [v1, v0] `shouldBe` simple res+    it "logical scoping over a parent" $ do+      let v0 = myv "v0" [] []+      let v1 = myv "v1" [v0] []+      let v2 = myv "v2" [v0] [v1]+      let res = [ (Nothing, ["v0", "v2"])+                , (Just "v0", [])+                , (Just "v1", [])+                , (Just "v2", ["v1"]) ]+      buildScopes [v2] `shouldBe` simple res+    it "common ancestor" $ do+      let top = myv "top" [] []+      let inner = myv "inner" [top] []+      let v1 = myv "v1" [top] [inner]+      let v2 = myv "v2" [top] [inner]+      let res = [ (Nothing, ["top", "v1", "v2"])+                , (Just "inner", [])+                , (Just "top", [])+                , (Just "v1", ["inner"])+                , (Just "v2", ["inner"]) ]+      buildScopes [v1, v2] `shouldBe` simple res+    it "common ancestor, unbalanced" $ do+      let top = myv "top" [] []+      let inner = myv "inner" [top] []+      let v1 = myv "v1" [top] [inner]+      let v2 = myv "v2" [] [inner]+      let res = [ (Nothing, ["top", "v1", "v2"])+                , (Just "inner", [])+                , (Just "top", [])+                , (Just "v1", ["inner"])+                , (Just "v2", ["inner", "top"]) ]+      buildScopes [v1, v2] `shouldBe` simple res+  describe "Path gatherings" $ do+    it "nothing" $ do+      gatherPaths' [] `shouldBe` gatherings []+    it "no parent" $ do+      let v0 = myv "v0" [] []+      let res = [("v0", [[]])]+      gatherPaths' [v0] `shouldBe` gatherings res+    it "one logical parent" $ do+      let v0 = myv "v0" [] []+      let v1 = myv "v1" [v0] []+      let res = [ ("v1", [[]])+                , ("v0", [[]])]+      gatherPaths' [v1] `shouldBe` gatherings res+    it "one inner parent" $ do+      let v0 = myv "v0" [] []+      let v1 = myv "v1" [] [v0]+      let res = [ ("v1", [[]])+                , ("v0", [["v1"]])]+      gatherPaths' [v1] `shouldBe` gatherings res+    it "logical scoping over a parent" $ do+      let v0 = myv "v0" [] []+      let v1 = myv "v1" [v0] []+      let v2 = myv "v2" [v0] [v1]+      let res = [ ("v0", [[]])+                , ("v1", [["v2"]])+                , ("v2", [[]]) ]+      gatherPaths' [v2] `shouldBe` gatherings res+    it "common ancestor" $ do+      let top = myv "top" [] []+      let inner = myv "inner" [top] []+      let v1 = myv "v1" [top] [inner]+      let v2 = myv "v2" [top] [inner]+      let res = [ ("inner", [["v1"], ["v2"]])+                , ("top", [[]])+                , ("v1", [[]])+                , ("v2", [[]]) ]+      gatherPaths' [v1, v2] `shouldBe` gatherings res+  describe "Real paths" $ do+    it "simple test" $ do+      let c0 = constant (1 :: Int) @@ "c0"+      let c1 = identity c0 @@ "c1"+      let c2 = identity c1 `logicalParents` [untyped c0] @@ "c2"+      nodeId <$> nodeParents c1 `shouldBe` [nodeId c0]+      nodeId <$> nodeParents c2 `shouldBe` [nodeId c1]+      let withParents = T.unpack . catNodePath . nodePath <$> assignPaths (untyped c2)+      withParents `shouldBe` ["c0", "c2/c1", "c2"]+    it "simple test 2" $ do+      let ds = dataset ([1 ,2, 3, 4]::[Int]) @@ "ds"+      let c = count ds @@ "c"+      let c2 = (c + (identity c @@ "id")) `logicalParents` [untyped ds] @@ "c2"+      let withParents = T.unpack . catNodePath . nodePath <$> assignPaths (untyped c2)+      withParents `shouldBe`  ["ds","c2/select_44dd5f","c2/c","c2/id","c2"]
+ test/Spark/Core/Internal/RowUtilsSpec.hs view
@@ -0,0 +1,32 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Spark.Core.Internal.RowUtilsSpec where++import Data.Aeson+import Data.Maybe(fromJust)+import Test.Hspec+import Data.ByteString.Lazy(ByteString)+import qualified Data.Vector as V++import Spark.Core.Types+import Spark.Core.Row+import Spark.Core.Internal.TypesFunctions++fun :: ByteString -> DataType -> Cell -> IO ()+fun js dt cell2 =+  let+    mval = decode js :: Maybe Value+    val = fromJust mval+    cellt = jsonToCell dt val+  in cellt `shouldBe` (Right cell2)+++spec :: Spec+spec = do+  describe "JSON -> Row" $ do+    it "ints" $ do+      fun "2" intType (IntElement 2)+    it "[ints]" $ do+      fun "[2]" (arrayType' intType) (RowArray (V.singleton (IntElement 2)))
+ test/Spark/Core/PathSpec.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleInstances #-}++module Spark.Core.PathSpec where++import Data.Maybe(fromJust)+import Test.Hspec++import Spark.Core.Functions+import Spark.Core.Dataset++fun1 :: LocalData Int+fun1 =+  let m1 = constant (1::Int)+      m2 = constant (2::Int) in+    constant (3::Int)+      `logicalParents` [untyped m1, untyped m2]+      `parents` []+++fun2 :: LocalData Int -> LocalData Int+fun2 ld1 = let+  m1 = constant (1 :: Int) `parents` [untyped ld1] @@ "m1"+    in+      constant (3 :: Int)+        `logicalParents` [untyped m1]+        `parents` [untyped ld1]+        @@ "c2"++-- fun3 :: LocalData Int -> LocalData Int+-- fun3 ld = ld + 3++spec :: Spec+spec = do+  describe "Tests with nodes" $ do+    it "should get a node" $ do+      let n1 = fun1+      let l = fromJust $ nodeLogicalParents n1+      (length l) `shouldBe` 2+    -- it "should work with ints" $ do+    --   let n2 = (fun3 4) @@ "" in+    --     (length $ nodeDependencies n2) `shouldBe` 2
+ test/Spark/Core/ProjectionsSpec.hs view
@@ -0,0 +1,112 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}++module Spark.Core.ProjectionsSpec where++import Test.Hspec+import GHC.Generics+import Data.List(isPrefixOf)+import Data.Either(isRight, isLeft)+import qualified Data.Vector as V+import qualified Data.Text as T++import Spark.Core.Functions+import Spark.Core.Dataset+import Spark.Core.Column+import Spark.Core.Row+import Spark.Core.Types+import Spark.Core.Try+import Spark.Core.Internal.Utilities+import Spark.Core.Internal.TypesFunctions+++data Tree = Tree {+  treeId :: Int,+  treeWidth :: Int,+  treeHeight :: Int } deriving (Generic, Show)++treeId' :: StaticColProjection Tree Int+treeId' = unsafeStaticProjection buildType "treeId"+treeWidth' :: StaticColProjection Tree Int+treeWidth' = unsafeStaticProjection buildType "treeWidth"+instance SQLTypeable Tree+instance ToSQL Tree++newtype MyId = MyId Int deriving (Generic, Show, Num)+instance SQLTypeable MyId+instance ToSQL MyId++newtype Height = Height Int deriving (Generic, Num, Show)+instance SQLTypeable Height+instance ToSQL Height++data STree = STree {+  sTreeId :: MyId,+  sTreeWidth :: Height,+  sTreeHeight :: Int } deriving (Generic, Show)++instance SQLTypeable STree+instance ToSQL STree+sTreeId' :: StaticColProjection STree MyId+sTreeId' = unsafeStaticProjection buildType "sTreeId"+sTreeWidth' :: StaticColProjection STree Height+sTreeWidth' = unsafeStaticProjection buildType "sTreeWidth"+instance TupleEquivalence STree (MyId, Height, Int) where+  tupleFieldNames = NameTuple ["sTreeId", "sTreeWidth", "sTreeHeight"]++rawData :: [(Int, Int, Int)]+rawData = [(1, 3, 2)]++spec :: Spec+spec = do+  let ds = dataset [Tree 1 3 2]+  -- The untyped elements+  let dt = structType [structField (T.pack "treeId") intType, structField (T.pack "treeWidth") intType, structField (T.pack "treeHeight") intType]+  let fun (id', height, width) = RowArray $ V.fromList [IntElement id', IntElement height, IntElement width]+  let df1 = traceHint (T.pack "df1=") $ dataframe dt (fun <$> rawData)+  let ds1 = traceHint (T.pack "ds1=") $ forceRight (asDS df1) :: Dataset Tree+  describe "Simple projection demos" $ do+    it "should get a node" $ do+      ds `shouldBe` ds1+    it "Failing dynamic projection on dataframe" $ do+      df1//"xx" `shouldSatisfy` isLeft+    it "Failing dynamic projection on dataset" $ do+      ds1//"xx" `shouldSatisfy` isLeft+    it "Basic arithmetic on DS cols" $ do+      let c1 = ds1//treeWidth'+      let c2 = (c1 + c1)+      (show c2) `shouldSatisfy` ("sum(treeWidth, treeWidth){int}" `isPrefixOf`)+    it "Basic arithmetic on DF cols" $ do+      let c1 = df1//treeWidth'+      let c2 = c1 + c1+      (show c2) `shouldSatisfy` ("Right sum(treeWidth, treeWidth){int}" `isPrefixOf`)+    it "Construction of ds2" $ do+      let str = struct' [ (df1//"treeId") @@ "sTreeId",+                          (df1//"treeWidth") @@ "sTreeWidth",+                          (df1//"treeHeight") @@ "sTreeHeight"]+      let df2 = pack' str+      let ds2 = traceHint (T.pack "ds2=") $ asDS df2 :: Try (Dataset STree)+      ds2 `shouldSatisfy` isRight+    it "Static construction of ds2" $ do+      let ds2 = do+              idCol <- castCol' (buildType::SQLType MyId) (df1//"treeId")+              widthCol <- castCol' (buildType::SQLType Height) (df1//"treeWidth")+              heightCol <- castCol' (buildType::SQLType Int) (df1//"treeWidth")+              let s = pack (idCol, widthCol, heightCol) :: Dataset STree+              return $ traceHint (T.pack "ds2=") s+      ds2 `shouldSatisfy` isRight+    it "Basic arithmetic on DS cols 1" $ do+      let ds2' = do+              idCol <- castCol' (buildType::SQLType MyId) (df1//"treeId")+              widthCol <- castCol' (buildType::SQLType Height) (df1//"treeWidth")+              heightCol <- castCol' (buildType::SQLType Int) (df1//"treeWidth")+              let s = pack (idCol, widthCol, heightCol) :: Dataset STree+              return $ traceHint (T.pack "ds2=") s+      let ds2 = forceRight ds2'+      let c1 = ds2//sTreeWidth'+      let c2 = c1 + c1+      (show c2) `shouldSatisfy` ("sum(sTreeWidth, sTreeWidth){int}" `isPrefixOf`)
+ test/Spark/Core/RowToSQLSpec.hs view
@@ -0,0 +1,68 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE BangPatterns #-}++module Spark.Core.RowToSQLSpec where++import qualified Data.Vector as V+import GHC.Generics (Generic)+import Test.Hspec++import Spark.Core.Types+import Spark.Core.Row++data TestStruct1 = TestStruct1 {+  ts1f1 :: Int,+  ts1f2 :: Maybe Int } deriving (Show, Eq, Generic, ToSQL, FromSQL)++data TestStruct2 = TestStruct2 { ts2f1 :: [Int] } deriving (Show, Generic, SQLTypeable)++data TestStruct3 = TestStruct3 { ts3f1 :: Int } deriving (Show, Eq, Generic, ToSQL, FromSQL)+data TestStruct4 = TestStruct4 { ts4f1 :: TestStruct3 } deriving (Show, Eq, Generic, ToSQL, FromSQL)++data TestStruct5 = TestStruct5 {+  ts5f1 :: Int,+  ts5f2 :: Int,+  ts5f3 :: TestStruct3+} deriving (Show, Eq, Generic, ToSQL, FromSQL)++newtype TestT1 = TestT1 { unTestT1 :: Int } deriving (Eq, Show, Generic, ToSQL, FromSQL)+++v2c :: (Show a, ToSQL a, FromSQL a, Eq a) => a -> Cell -> IO ()+v2c !x !y = do+  _ <- shouldBe (valueToCell x) y+  _ <- shouldBe (cellToValue y) (Right x)+  return ()++spec :: Spec+spec = do+  describe "Simple type tests" $ do+    it "int" $+      v2c (3 :: Int) (IntElement 3)+    it "int?" $+      v2c (Just 3 :: Maybe Int) (IntElement 3)+    it "int? 2" $+      v2c (Nothing :: Maybe Int) Empty+    it "TestStruct3" $+      v2c (TestStruct3 2) (RowArray $ V.fromList [IntElement 2])+    it "TestStruct4" $+      v2c (TestStruct4 (TestStruct3 3)) $+        (RowArray $ V.fromList [+            RowArray $ V.fromList [IntElement 3]+          ])+    it "TestStruct1 - empty" $+      v2c (TestStruct1 2 Nothing) (RowArray $ V.fromList [IntElement 2, Empty])+    it "TestStruct1 - full" $+      v2c (TestStruct1 2 (Just 4)) (RowArray $ V.fromList [IntElement 2, IntElement 4])+    it "TestStruct5" $+      v2c (TestStruct5 1 2 (TestStruct3 3)) $+        (RowArray $ V.fromList [+            IntElement 1,+            IntElement 2,+            RowArray $ V.fromList [IntElement 3]+          ])+  -- describe "Simple type tests" $ do+  --   it "newtype" $+  --     v2c (TestT1 3) (IntElement 3)
+ test/Spark/Core/SimpleExamplesSpec.hs view
@@ -0,0 +1,75 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE QuasiQuotes #-}++-- Some small examples that get fully verified.+module Spark.Core.SimpleExamplesSpec where++import Data.Either(isRight)+import Data.Maybe(isJust)+import Test.Hspec+import qualified Data.Text as T+import Text.RawString.QQ++import Spark.Core.Dataset+import Spark.Core.Functions+import Spark.Core.Column+import Spark.Core.ColumnFunctions+import Spark.Core.Internal.DatasetStructures+import Spark.Core.Internal.Utilities(pretty)+import Spark.Core.Internal.OpFunctions(extraNodeOpData)++ds1 :: Dataset Int+ds1 = dataset [1,2,3]++ds2 :: Dataset Double+ds2 = error "ds2"++spec :: Spec+spec = do+  describe "Simple examples" $ do+    it "Precdence of renaming" $ do+      let numbers = asCol ds1+      let s = colSum numbers+      let numCount = count ds1+      let avg = s `div` numCount @@ "myaverage"+      _cnName avg `shouldSatisfy` isJust+    it "name for simple integers" $ do+      let numbers = asCol ds1+      let s = colSum numbers+      let numCount = count ds1+      let avg = s `div` numCount @@ "myaverage"+      -- TODO: should it show "value: int" instead?+      -- I think it should show it for distributed nodes only.+      -- SQL is not allowed on observables+      (show avg) `shouldBe` "myaverage@org.spark.LocalDiv!{int}"+  describe "pack1" $ do+    it "Extracting and packing one column" $ do+      let numbers = asCol ds1+      let ds1' = pack1 numbers+      (nodeType ds1) `shouldBe` (nodeType ds1')+  describe "pack" $ do+    it "Extracting and packing one column" $ do+      let ds1' = pack' . asCol $ ds1+      (nodeType <$> (asDF ds1)) `shouldBe` (nodeType <$> ds1')+  describe "simple json example" $ do+    it "packing and unpacking one column" $ do+      let ds1' = pack' . asCol $ ds1+      let d' = pretty . extraNodeOpData . nodeOp <$> ds1'+      d' `shouldBe` Right (T.pack [r|{"colOp":"extraction","field":[]}|])+    it "packing and unpacking 2 columns, one with a bad name" $ do+      let col1 = asCol ds1+      let col2 = col1 @@ "other"+      let ds1' = pack' (col1, col2)+      ds1' `shouldSatisfy` isRight -- NOT SURE WHY IT WOULD FAIL+    it "packing and unpacking 2 columns, one with a good name" $ do+      let col1 = asCol ds1 @@ "first"+      let col2 = col1 @@ "second"+      let ds1' = pack' (col1, col2)+      ds1' `shouldSatisfy` isRight+++    -- it "example2" $ do+    --   let numbers = asCol ds2+    --   let avg = (colSum numbers) / (count ds2)+    --   1 `shouldBe` 1
+ test/Spark/Core/TypesSpec.hs view
@@ -0,0 +1,95 @@+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE OverloadedStrings #-}++module Spark.Core.TypesSpec where++import GHC.Generics (Generic)+import Test.Hspec+import Test.Hspec.QuickCheck++import Spark.Core.Types+import Spark.Core.Internal.TypesFunctions++data TestStruct1 = TestStruct1 {+  ts1f1 :: Int,+  ts1f2 :: Maybe Int } deriving (Show, Generic, SQLTypeable)++data TestStruct2 = TestStruct2 { ts2f1 :: [Int] } deriving (Show, Generic, SQLTypeable)++data TestStruct3 = TestStruct3 { ts3f1 :: Int } deriving (Show, Generic, SQLTypeable)+data TestStruct4 = TestStruct4 { ts4f1 :: TestStruct3 } deriving (Show, Generic, SQLTypeable)++-- instance SQLTypeable TestStruct1+-- instance Menu TestStruct1+++-- main :: IO ()+-- main = hspec spec++spec :: Spec+spec = do+  describe "Simple type tests" $ do+    it "show ints" $+      show intType `shouldBe` "int"++    it "show arrays" $+      show (arrayType' intType) `shouldBe` "[int]"++    it "show structures" $+      show (arrayType' (canNull intType)) `shouldBe` "[int?]"++  describe "The basic tests for int types" $ do+    it "ints" $+      let t = buildType :: (SQLType Int)+          dt = columnType t in+        dt `shouldBe` intType++    it "opt ints" $+      let t = buildType :: (SQLType (Maybe Int))+          dt = columnType t in+        dt `shouldBe` canNull intType++    -- The projection of all the product types+    it "opt opt ints" $+      let t = buildType :: (SQLType (Maybe (Maybe Int))) in+        columnType t `shouldBe` canNull intType++    it "array ints" $+      let t = buildType :: (SQLType [Int]) in+        columnType t `shouldBe` arrayType' intType++    it "array opt ints" $+      let t = buildType :: (SQLType [Maybe Int]) in+        columnType t `shouldBe` arrayType' (canNull intType)++    it "opt array ints" $+      let t = buildType :: (SQLType (Maybe [Int])) in+        columnType t `shouldBe` canNull (arrayType' intType)++  describe "The basic tests for records" $ do+    it "records with maybe" $+      let t = buildType :: (SQLType TestStruct1)+          out = structType [structField "ts1f1" intType, structField "ts1f2" (canNull intType)] in+        columnType t `shouldBe` out++    it "records with arrays" $+      let t = buildType :: (SQLType TestStruct2)+          out = structType [structField "ts2f1" (arrayType' intType)] in+        columnType t `shouldBe` out++    it "records within records" $+      let t = buildType :: (SQLType TestStruct4)+          out0 = structType [structField "ts3f1" intType]+          out = structType [structField "ts4f1" out0] in+        columnType t `shouldBe` out++  describe "Construction of frame types" $ do+    prop "frameTypeFromCol should be invertible" $+      \x ->+        let dt = colTypeFromFrame x+            y = frameTypeFromCol dt+        in x == y+    -- TODO this is not always working. Figure out the rules here.+    -- prop "colTypeFromFrame should be invertible" $+    --   \x -> (colTypeFromFrame . frameTypeFromCol) x == x
+ test/Spec.hs view
@@ -0,0 +1,2 @@+-- Not working???+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}