Add Pipeline.execute_streaming() method

Author: gsakkis

README.rst

@@ -187,7 +187,7 @@ PDAL and Python:
     ).pipeline(clamped)
     print(pipeline.execute())  # 387 points
 
-Iterating Streamable Pipelines
+Executing Streamable Pipelines
 ................................................................................
 Streamable pipelines (pipelines that consist exclusively of streamable PDAL
 stages) can be executed in streaming mode via ``Pipeline.iterator()``. This
@@ -207,6 +207,12 @@ returns an iterator object that yields Numpy arrays of up to ``chunk_size`` size
 to allow prefetching up to to this number of arrays in parallel and buffering
 them until they are yielded to the caller.
 
+If you just want to execute a streamable pipeline in streaming mode and don't
+need to access the data points (typically when the pipeline has Writer stage(s)),
+you can use the ``Pipeline.execute_streaming(chunk_size)`` method instead. This
+is functionally equivalent to ``sum(map(len, pipeline.iterator(chunk_size)))``
+but more efficient as it avoids allocating and filling any arrays in memory.
+
 Accessing Mesh Data
 ................................................................................
 

pdal/PyPipeline.cpp

@@ -44,6 +44,16 @@ namespace pdal
 namespace python
 {
 
+
+void CountPointTable::reset()
+{
+    for (PointId idx = 0; idx < numPoints(); idx++)
+        if (!skip(idx))
+            m_count++;
+    FixedPointTable::reset();
+}
+
+
 PipelineExecutor::PipelineExecutor(
     std::string const& json, std::vector<std::shared_ptr<Array>> arrays, int level)
 {
@@ -70,6 +80,14 @@ point_count_t PipelineExecutor::execute()
 }
 
 
+point_count_t PipelineExecutor::executeStream(point_count_t streamLimit)
+{
+    CountPointTable table(streamLimit);
+    m_manager.executeStream(table);
+    m_executed = true;
+    return table.count();
+}
+
 const PointViewSet& PipelineExecutor::views() const
 {
     if (!m_executed)

pdal/PyPipeline.hpp

@@ -54,6 +54,7 @@ class PDAL_DLL PipelineExecutor {
     virtual ~PipelineExecutor() = default;
 
     point_count_t execute();
+    point_count_t executeStream(point_count_t streamLimit);
 
     const PointViewSet& views() const;
     std::string getPipeline() const;
@@ -73,5 +74,18 @@ class PDAL_DLL PipelineExecutor {
     std::stringstream m_logStream;
 };
 
+class CountPointTable : public FixedPointTable
+{
+public:
+    CountPointTable(point_count_t capacity) : FixedPointTable(capacity), m_count(0) {}
+    point_count_t count() const { return m_count; }
+
+protected:
+    virtual void reset();
+
+private:
+    point_count_t m_count;
+};
+
 } // namespace python
 } // namespace pdal

pdal/libpdalpython.pyx

@@ -92,6 +92,7 @@ cdef extern from "PyPipeline.hpp" namespace "pdal::python":
     cdef cppclass PipelineExecutor:
         PipelineExecutor(string, vector[shared_ptr[Array]], int) except +
         int execute() except +
+        int executeStream(int) except +
         string getPipeline() except +
         string getMetadata() except +
         string getSchema() except +
@@ -185,6 +186,9 @@ cdef class Pipeline(PipelineResultsMixin):
     def execute(self):
         return self._get_executor().execute()
 
+    def execute_streaming(self, int chunk_size=10000):
+        return self._get_executor().executeStream(chunk_size)
+
     def iterator(self, int chunk_size=10000, int prefetch = 0):
         it = PipelineIterator()
         it.set_executor(new StreamableExecutor(

test/test_pipeline.py

@@ -56,12 +56,28 @@ def test_invalid_json(self, pipeline):
             pdal.Pipeline(pipeline)
 
     @pytest.mark.parametrize("filename", ["sort.json", "sort.py"])
-    def test_execution(self, filename):
+    def test_execute(self, filename):
         """Can we execute a PDAL pipeline"""
         r = get_pipeline(filename)
         r.execute()
         assert len(r.pipeline) > 200
 
+    @pytest.mark.parametrize("filename", ["range.json", "range.py"])
+    def test_execute_streaming(self, filename):
+        r = get_pipeline(filename)
+        assert r.streamable
+        count = r.execute()
+        count2 = r.execute_streaming(chunk_size=100)
+        assert count == count2
+
+    @pytest.mark.parametrize("filename", ["sort.json", "sort.py"])
+    def test_execute_streaming_non_streamable(self, filename):
+        r = get_pipeline(filename)
+        assert not r.streamable
+        with pytest.raises(RuntimeError) as info:
+            r.execute_streaming()
+        assert "Attempting to use stream mode" in str(info.value)
+
     @pytest.mark.parametrize("filename", ["bad.json", "bad.py"])
     def test_validate(self, filename):
         """Do we complain with bad pipelines"""