tuplex · LeonhardFS · Jan 24, 2022 · Nov 3, 2021 · Nov 3, 2021 · Nov 3, 2021
diff --git a/tuplex/awslambda/src/lambda_main.cc b/tuplex/awslambda/src/lambda_main.cc
@@ -329,7 +329,7 @@ tuplex::messages::InvocationResponse lambda_main(aws::lambda_runtime::invocation
     // parse protobuf request
     tuplex::messages::InvocationRequest req;
     auto rc = google::protobuf::util::JsonStringToMessage(lambda_req.payload, &req);
-    if(rc != google::protobuf::util::Status::OK)
+    if(!rc.ok())
         throw std::runtime_error("could not parse json into protobuf message, bad parse for request");
 
     timer.mark_time("decode_request");

diff --git a/tuplex/core/include/Context.h b/tuplex/core/include/Context.h
@@ -37,6 +37,7 @@ namespace tuplex {
     class Executor;
     class Partition;
     class IBackend;
+    class ExceptionInfo;
 
     class Context {
     private:
@@ -58,7 +59,25 @@ namespace tuplex {
 
         // needed because of C++ template issues
         void addPartition(DataSet* ds, Partition *partition);
-        void addParallelizeNode(DataSet *ds); //! adds a paralellize node to the computation graph
+        void addParallelizeNode(DataSet *ds, const std::vector<std::tuple<size_t, PyObject*>> &badParallelizeObjects=std::vector<std::tuple<size_t, PyObject*>>(), const std::vector<size_t> &numExceptionsInPartition=std::vector<size_t>()); //! adds a paralellize node to the computation graph
+
+        /*!
+         * serialize python objects as pickled objects into exception partitions. Set the python objects map to
+         * map all normalPartitions to the exceptions that occured within them.
+         * @param pythonObjects normal case schema violations and their initial row numbers
+         * @param numExceptionsInPartition number of exceptions in each normal partition
+         * @param normalPartitions normal partitions created
+         * @param opID parallelize operator ID
+         * @param serializedPythonObjects output vector for partitions
+         * @param pythonObjectsMap output for mapping
+         */
+        void serializePythonObjects(const std::vector<std::tuple<size_t, PyObject*>>& pythonObjects,
+                                    const std::vector<size_t> &numExceptionsInPartition,
+                                    const std::vector<Partition*> &normalPartitions,
+                                    const int64_t opID,
+                                    std::vector<Partition*> &serializedPythonObjects,
+                                    std::unordered_map<std::string, ExceptionInfo> &pythonObjectsMap);
+
         Partition* requestNewPartition(const Schema& schema, const int dataSetID, size_t minBytesRequired);
         uint8_t* partitionLockRaw(Partition *partition);
         void partitionUnlock(Partition *partition);
@@ -111,7 +130,8 @@ namespace tuplex {
                                             columnNames);
         }
 
-        DataSet& parallelize(const std::vector<Row>& L, const std::vector<std::string>& columnNames=std::vector<std::string>());
+        DataSet& parallelize(const std::vector<Row>& L,
+                             const std::vector<std::string>& columnNames=std::vector<std::string>());
 
         DataSet& parallelize(std::initializer_list<double> L, const std::vector<std::string>& columnNames=std::vector<std::string>()) {
             if(!columnNames.empty())
@@ -252,9 +272,11 @@ namespace tuplex {
          *        empty dataset will be created.
          *
          * @param columns optional column names
+         * @param badParallelizeObjects schema violations found during parallelization of partitions
+         * @param numExceptionsInPartition number of schema violations that occured in each of the partitions
          * @return reference to newly created dataset.
          */
-        DataSet& fromPartitions(const Schema& schema, const std::vector<Partition*>& partitions, const std::vector<std::string>& columns);
+        DataSet& fromPartitions(const Schema& schema, const std::vector<Partition*>& partitions, const std::vector<std::string>& columns, const std::vector<std::tuple<size_t, PyObject*>> &badParallelizeObjects, const std::vector<size_t> &numExceptionsInPartition);
     };
     // needed for template mechanism to work
 #include <DataSet.h>

diff --git a/tuplex/core/include/ExceptionInfo.h b/tuplex/core/include/ExceptionInfo.h
@@ -0,0 +1,47 @@
+//--------------------------------------------------------------------------------------------------------------------//
+//                                                                                                                    //
+//                                      Tuplex: Blazing Fast Python Data Science                                      //
+//                                                                                                                    //
+//                                                                                                                    //
+//  (c) 2017 - 2021, Tuplex team                                                                                      //
+//  Created by Benjamin Givertz first on 1/1/2022                                                                     //
+//  License: Apache 2.0                                                                                               //
+//--------------------------------------------------------------------------------------------------------------------//
+
+#ifndef TUPLEX_EXCEPTIONINFO_H
+#define TUPLEX_EXCEPTIONINFO_H
+
+namespace tuplex {
+    /*!
+     * Struct to hold information that maps input partitions to input exceptions that occur within them.
+     *
+     * Explanation:
+     * Each input partition is passed the same vector of all input exceptions that occured during data parallelization
+     * or caching. Thus, each input partition must know how many input exceptions occur in its partition, the index
+     * of the input exception partition where its first exception occurs, and the offset into that partition where the
+     * first exception occurs. These values are held in this struct and each input partition is mapped to an ExceptionInfo.
+     */
+    struct ExceptionInfo {
+        size_t numExceptions; //! number of exception rows that occur within a single input partition
+        size_t exceptionIndex; //! index into a vector of input exception partitions that holds the first input exception
+        size_t exceptionRowOffset; //! offset in rows into the first input exception partition where the first exception occurs.
+        size_t exceptionByteOffset; //! offset in bytes into the first input exception partition where the first exception occurs
+
+        ExceptionInfo() :
+                numExceptions(0),
+                exceptionIndex(0),
+                exceptionRowOffset(0),
+                exceptionByteOffset(0) {}
+
+        ExceptionInfo(size_t numExps,
+                      size_t expIndex,
+                      size_t expRowOffset,
+                      size_t expByteOffset) :
+                numExceptions(numExps),
+                exceptionIndex(expIndex),
+                exceptionRowOffset(expRowOffset),
+                exceptionByteOffset(expByteOffset) {}
+    };
+}
+
+#endif //TUPLEX_EXCEPTIONINFO_H
diff --git a/tuplex/core/include/ee/local/LocalBackend.h b/tuplex/core/include/ee/local/LocalBackend.h
@@ -66,7 +66,7 @@ namespace tuplex {
         void freeExecutors();
 
 
-        std::vector<IExecutorTask*> createLoadAndTransformToMemoryTasks(TransformStage* tstage, const ContextOptions& options,  codegen::read_block_f functor);
+        std::vector<IExecutorTask*> createLoadAndTransformToMemoryTasks(TransformStage* tstage, const ContextOptions& options, const std::shared_ptr<TransformStage::JITSymbols>& syms);
         void executeTransformStage(TransformStage* tstage);
 
 

diff --git a/tuplex/core/include/logical/CacheOperator.h b/tuplex/core/include/logical/CacheOperator.h
@@ -81,6 +81,8 @@ namespace tuplex {
         bool isCached() const { return _cached; }
         std::vector<Partition*> cachedPartitions() const { return _normalCasePartitions; }
         std::vector<Partition*> cachedExceptions() const { return _generalCasePartitions; }
+        std::unordered_map<std::string, ExceptionInfo> partitionToExceptionsMap() const { return _partitionToExceptionsMap; }
+
         size_t getTotalCachedRows() const;
 
         int64_t cost() const override;
@@ -109,7 +111,8 @@ namespace tuplex {
         std::vector<Partition*> _generalCasePartitions;   //! holds all data which is considered to be a normal-case violation,
                                                           //! i.e. which does not adhere to the normal case schema, but did not produce
                                                           //! an exception while being processed through the pipeline before
-        std::vector<PyObject*>  _py_objects;              //! all python objects who do not adhere to the general case schema (
+        std::unordered_map<std::string, ExceptionInfo> _partitionToExceptionsMap; //! maps normal case partitions to corresponding general case ones
+        std::vector<PyObject*>  _py_objects;              //! all python objects who do not adhere to the general case schema
         std::vector<std::string> _columns;
 
         // internal sample of normal case rows, used for tracing & Co.

diff --git a/tuplex/core/include/logical/ParallelizeOperator.h b/tuplex/core/include/logical/ParallelizeOperator.h
@@ -18,7 +18,9 @@ namespace tuplex {
     class ParallelizeOperator : public LogicalOperator {
 
         std::vector<Partition*> _partitions; // data, conforming to majority type
-        //@TODO: missing: python objects & general case data
+        std::vector<Partition*> _pythonObjects; // schema violations stored for interpreter processing as python objects
+        // maps partitions to their corresponding python objects
+        std::unordered_map<std::string, ExceptionInfo> _inputPartitionToPythonObjectsMap;
         std::vector<std::string> _columnNames;
 
         std::vector<Row> _sample; // sample, not necessary conforming to one type
@@ -28,7 +30,9 @@ namespace tuplex {
         LogicalOperator *clone() override;
 
         // this a root node
-        ParallelizeOperator(const Schema& schema, std::vector<Partition*> partitions, const std::vector<std::string>& columns);
+        ParallelizeOperator(const Schema& schema,
+                            const std::vector<Partition*>& partitions,
+                            const std::vector<std::string>& columns);
 
         std::string name() override { return "parallelize"; }
         LogicalOperatorType type() const override { return LogicalOperatorType::PARALLELIZE; }
@@ -45,6 +49,12 @@ namespace tuplex {
          */
         std::vector<tuplex::Partition*> getPartitions();
 
+        void setPythonObjects(const std::vector<Partition*> &pythonObjects) { _pythonObjects = pythonObjects; }
+        std::vector<Partition *> getPythonObjects() { return _pythonObjects; }
+
+        void setInputPartitionToPythonObjectsMap(const std::unordered_map<std::string, ExceptionInfo>& pythonObjectsMap) { _inputPartitionToPythonObjectsMap = pythonObjectsMap; }
+        std::unordered_map<std::string, ExceptionInfo> getInputPartitionToPythonObjectsMap() { return _inputPartitionToPythonObjectsMap; }
+
         Schema getInputSchema() const override { return getOutputSchema(); }
 
         bool good() const override;

diff --git a/tuplex/core/include/physical/BlockBasedTaskBuilder.h b/tuplex/core/include/physical/BlockBasedTaskBuilder.h
@@ -30,6 +30,13 @@ namespace tuplex {
 
             llvm::Function *createFunction();
 
+            /*!
+             * create function to process blocks of data along with any input exceptions. Used when filters are present
+             * to update the indices of the exceptions.
+             * @return llvm function
+             */
+            llvm::Function *createFunctionWithExceptions();
+
             python::Type _inputRowType; //@TODO: make this private??
 
             std::string _intermediateCallbackName;
@@ -48,7 +55,6 @@ namespace tuplex {
              * @return
              */
             inline llvm::Value *arg(const std::string &name) {
-                assert(_args.size() == 6);
                 auto it = _args.find(name);
                 if (it == _args.end())
                     throw std::runtime_error("unknown arg " + name + " requested");

diff --git a/tuplex/core/include/physical/CodeDefs.h b/tuplex/core/include/physical/CodeDefs.h
@@ -47,6 +47,10 @@ namespace tuplex {
         // parameters are userData, array of globals (e.g. hashmaps), block, blocksize, normalrowsout, badrowsout, lastRow
         typedef int64_t(*read_block_f)(void*, const uint8_t*, int64_t, int64_t*, int64_t*, int8_t);
 
+        // protoype of the function generated by the below builder
+        // parameters are userData, block, blocksize, expPtrs, expPtrSizes, numExceptions, normalrowsout, badrowsout, lastRow
+        typedef int64_t(*read_block_exp_f)(void*, const uint8_t*, int64_t, uint8_t **, int64_t *, int64_t, int64_t*, int64_t*, bool);
+
         /*!
          * prototype for processing a single row (with callbacks etc.). Returns how many bytes were processed
          */

diff --git a/tuplex/core/include/physical/ExceptionSourceTaskBuilder.h b/tuplex/core/include/physical/ExceptionSourceTaskBuilder.h
@@ -0,0 +1,63 @@
+//--------------------------------------------------------------------------------------------------------------------//
+//                                                                                                                    //
+//                                      Tuplex: Blazing Fast Python Data Science                                      //
+//                                                                                                                    //
+//                                                                                                                    //
+//  (c) 2017 - 2021, Tuplex team                                                                                      //
+//  Created by Benjamin Givertz first on 1/1/2021                                                                     //
+//  License: Apache 2.0                                                                                               //
+//--------------------------------------------------------------------------------------------------------------------//
+
+#ifndef TUPLEX_EXCEPTIONSOURCETASKBUILDER_H
+#define TUPLEX_EXCEPTIONSOURCETASKBUILDER_H
+
+#include "BlockBasedTaskBuilder.h"
+
+namespace tuplex {
+    namespace codegen {
+        /*!
+         * Class used to process tuplex partitions through the pipeline when both input exceptions and filters are
+         * present. This is necessary to update the row indices of any input exceptions if rows are filtered out.
+         */
+        class ExceptionSourceTaskBuilder : public BlockBasedTaskBuilder {
+        private:
+            void createMainLoop(llvm::Function* read_block_func, bool terminateEarlyOnLimitCode);
+
+            /*!
+            * generates code to process a row depending on parse result...
+            * @param builder
+            * @param userData a value for userData (i.e. the class ptr of the task typically) to be parsed to callback functions
+            * @param tuple (flattened) tuple representation of current tuple (LLVM)
+            * @param normalRowCountVar where to store normal row counts
+            * @param badRowCountVar where to store bad row counts
+            * @param processRowFunc (optional) function to be called before output is written.
+            *        Most likely this is not a nullptr, because users want to transform data.
+            */
+            void processRow(llvm::IRBuilder<>& builder,
+                            llvm::Value* userData,
+                            const FlattenedTuple& tuple,
+                            llvm::Value *normalRowCountVar,
+                            llvm::Value *badRowCountVar,
+                            llvm::Value *outputRowNumberVar,
+                            llvm::Value *inputRowPtr,
+                            llvm::Value *inputRowSize,
+                            bool terminateEarlyOnLimitCode,
+                            llvm::Function* processRowFunc=nullptr);
+
+            void callProcessFuncWithHandler(llvm::IRBuilder<> &builder, llvm::Value *userData,
+                                            const FlattenedTuple &tuple,
+                                            llvm::Value *normalRowCountVar, llvm::Value *badRowCountVar, llvm::Value *rowNumberVar,
+                                            llvm::Value *inputRowPtr, llvm::Value *inputRowSize,
+                                            bool terminateEarlyOnLimitCode,
+                                            llvm::Function *processRowFunc);
+        public:
+            ExceptionSourceTaskBuilder() = delete;
+
+            explicit ExceptionSourceTaskBuilder(const std::shared_ptr<LLVMEnvironment>& env, const python::Type& rowType, const std::string& name) : BlockBasedTaskBuilder::BlockBasedTaskBuilder(env, rowType, name)   {}
+
+            llvm::Function* build(bool terminateEarlyOnFailureCode) override;
+        };
+    }
+}
+
+#endif //TUPLEX_EXCEPTIONSOURCETASKBUILDER_H