WIP Add HThorGenericDiskWriteActivity/BaseActivity

hpcc-systems · Sep 30, 2024 · a3be103 · a3be103
1 parent 6429f63
commit a3be103
Show file tree

Hide file tree

Showing 5 changed files with 103 additions and 1 deletion.
diff --git a/ecl/eclagent/eclgraph.cpp b/ecl/eclagent/eclgraph.cpp
@@ -53,7 +53,13 @@ static IHThorActivity * createActivity(IAgentContext & agent, unsigned activityI
     {
     case TAKdiskwrite:
     case TAKspillwrite:
-        return createDiskWriteActivity(agent, activityId, subgraphId, (IHThorDiskWriteArg &)arg, kind, graph);
+    {
+        // bool isGeneric = (((IHThorNewDiskReadArg &)arg).getFlags() & TDXgeneric) != 0;
+        if (1)
+            return createGenericDiskWriteActivity(agent, activityId, subgraphId, (IHThorNewDiskReadArg &)arg, kind, graph, node);
+        else
+            return createDiskWriteActivity(agent, activityId, subgraphId, (IHThorDiskWriteArg &)arg, kind, graph);
+    }
     case TAKsort:
         return createGroupSortActivity(agent, activityId, subgraphId, (IHThorSortArg &)arg, kind, graph);
     case TAKdedup:

diff --git a/ecl/hthor/hthor.cpp b/ecl/hthor/hthor.cpp
@@ -12003,6 +12003,55 @@ const void *CHThorGenericDiskReadActivity::nextRow()
     return NULL;
 }
 
+CHThorGenericDiskWriteBaseActivity::CHThorGenericDiskWriteBaseActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorNewDiskReadBaseArg &_arg, ThorActivityKind _kind, EclGraph & _graph, IPropertyTree *_node)
+: CHThorActivityBase(_agent, _activityId, _subgraphId, _arg, _kind, _graph), helper(_arg)
+{
+
+}
+
+void CHThorGenericDiskWriteBaseActivity::ready()
+{
+    // Implementation here
+}
+
+void CHThorGenericDiskWriteBaseActivity::stop()
+{
+    // Implementation here
+}
+
+void CHThorGenericDiskWriteBaseActivity::execute()
+{
+    // Implementation here
+}
+
+CHThorGenericDiskWriteActivity::CHThorGenericDiskWriteActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorNewDiskReadArg &_arg, ThorActivityKind _kind, EclGraph & _graph, IPropertyTree *_node)
+: CHThorGenericDiskWriteBaseActivity(_agent, _activityId, _subgraphId, _arg, _kind, _graph, _node), helper(_arg)
+{
+
+}
+
+// Implement all pure virtual functions from CHThorGenericDiskWriteBaseActivity
+void CHThorGenericDiskWriteActivity::ready()
+{
+    // Implementation here
+}
+
+void CHThorGenericDiskWriteActivity::stop()
+{
+    // Implementation here
+}
+
+void CHThorGenericDiskWriteActivity::execute()
+{
+    // Implementation here
+}
+
+const void *CHThorGenericDiskWriteActivity::nextRow()
+{
+    // Implementation here
+    return nullptr;
+}
+
 //=====================================================================================================
 
 MAKEFACTORY(DiskWrite);
@@ -12070,6 +12119,11 @@ extern HTHOR_API IHThorActivity *createGenericDiskReadActivity(IAgentContext &_a
     return new CHThorGenericDiskReadActivity(_agent, _activityId, _subgraphId, arg, kind, _graph, node);
 }
 
+extern HTHOR_API IHThorActivity *createGenericDiskWriteActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorNewDiskReadArg &arg, ThorActivityKind kind, EclGraph & _graph, IPropertyTree * node)
+{
+    return new CHThorGenericDiskWriteActivity(_agent, _activityId, _subgraphId, arg, kind, _graph, node);
+}
+
 MAKEFACTORY(Null);
 MAKEFACTORY(SideEffect);
 MAKEFACTORY(Action);

diff --git a/ecl/hthor/hthor.hpp b/ecl/hthor/hthor.hpp
@@ -174,6 +174,7 @@ extern HTHOR_API IHThorActivity *createDiskGroupAggregateActivity(IAgentContext
 
 extern HTHOR_API IHThorActivity *createNewDiskReadActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorNewDiskReadArg &arg, ThorActivityKind kind, EclGraph & _graph, IPropertyTree *node);
 extern HTHOR_API IHThorActivity *createGenericDiskReadActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorNewDiskReadArg &arg, ThorActivityKind kind, EclGraph & _graph, IPropertyTree * node);
+extern HTHOR_API IHThorActivity *createGenericDiskWriteActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorNewDiskReadArg &arg, ThorActivityKind kind, EclGraph & _graph, IPropertyTree * node);
 
 extern HTHOR_API IHThorActivity *createIndexReadActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorIndexReadArg &arg, ThorActivityKind kind, EclGraph & _graph, IPropertyTree *_node);
 extern HTHOR_API IHThorActivity *createIndexNormalizeActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorIndexNormalizeArg &arg, ThorActivityKind kind, EclGraph & _graph, IPropertyTree *_node);

diff --git a/ecl/hthor/hthor.ipp b/ecl/hthor/hthor.ipp
@@ -3241,6 +3241,43 @@ protected:
     void onLimitExceeded();
 };
 
+class CHThorGenericDiskWriteBaseActivity : public CHThorActivityBase/*, implements IThorDiskCallback, implements IIndexWriteContext, public IFileCollectionContext*/
+{
+protected:
+    IHThorNewDiskReadBaseArg &helper;
+    IDiskRowWriter * activeReader = nullptr;
+    CLogicalFileCollection files;
+    bool outputGrouped = false;
+public:
+    CHThorGenericDiskWriteBaseActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorNewDiskReadBaseArg &_arg, ThorActivityKind _kind, EclGraph & _graph, IPropertyTree *_node);
+    IMPLEMENT_IINTERFACE_USING(CHThorActivityBase)
+
+    virtual void ready();
+    virtual void stop();
+    virtual void execute();
+
+    //interface IHThorInput
+    virtual bool isGrouped()                                { return outputGrouped; }
+    virtual IOutputMetaData * queryOutputMeta() const       { return outputMeta; }
+};
+
+class CHThorGenericDiskWriteActivity : public CHThorGenericDiskWriteBaseActivity
+{
+    typedef CHThorGenericDiskWriteBaseActivity PARENT;
+protected:
+    IHThorNewDiskReadArg &helper;
+public:
+    CHThorGenericDiskWriteActivity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThorNewDiskReadArg &_arg, ThorActivityKind _kind, EclGraph & _graph, IPropertyTree *_node);
+
+    virtual void ready();
+    virtual void stop();
+    virtual void execute();
+    virtual bool needsAllocator() const { return true; }
+
+    //interface IHThorInput
+    virtual const void *nextRow();
+};
+
 
 #define MAKEFACTORY(NAME) \
 extern HTHOR_API IHThorActivity * create ## NAME ## Activity(IAgentContext &_agent, unsigned _activityId, unsigned _subgraphId, IHThor ## NAME ## Arg &arg, ThorActivityKind kind, EclGraph & _graph) \

diff --git a/testing/regress/ecl/genericActivities.ecl b/testing/regress/ecl/genericActivities.ecl
@@ -0,0 +1,4 @@
+OUTPUT(DATASET([{1}], {UNSIGNED1 a}), ,'data');
+
+d := DATASET('data', {UNSIGNED1 a}, TYPE(FLAT));
+OUTPUT(d);