Streamable EPT Reader (#2603)

io/EptReader.cpp

@@ -136,7 +136,7 @@ struct EptReader::Args
 };
 
 
-EptReader::EptReader() : m_args(new EptReader::Args)
+EptReader::EptReader() : m_args(new EptReader::Args), m_currentIndex(0)
 {}
 
 EptReader::~EptReader()
@@ -435,6 +435,11 @@ void EptReader::addDimensions(PointLayoutPtr layout)
     {
         throwError(e.what());
     }
+
+    // Backup the layout for streamable pipeline.
+    // Will be used to restore m_bufferPointTable layout after flushing points from previous tile.
+    if (pipelineStreamable())
+    	m_bufferLayout = layout;
 }
 
 void EptReader::ready(PointTableRef table)
@@ -770,5 +775,77 @@ Dimension::Type EptReader::getCoercedTypeTest(const NL::json& j)
     return getCoercedType(j);
 }
 
+void EptReader::loadNextOverlap()
+{
+    std::map<Key, uint64_t>::iterator itr = m_overlaps.begin();
+    std::advance(itr, m_nodeId-1);
+    Key key(itr->first);
+    log()->get(LogLevel::Debug)
+        << "Streaming Data " << m_nodeId << "/" << m_overlaps.size() << ": "
+        << key.toString() << std::endl;
+
+    uint64_t startId(0);
+
+    // Reset PointTable to make sure all points from previous tile are flushed.
+    m_bufferPointTable.reset(new PointTable());
+
+    // We did reset PointTable,
+    // So it will not have the dimensions registered to it.
+    // Register/Restore Dimensions for PointTable layout.
+    PointLayoutPtr layout = m_bufferPointTable->layout();
+    for (auto id : m_bufferLayout->dims())
+        layout->registerOrAssignDim(m_bufferLayout->dimName(id),
+                                    m_bufferLayout->dimType(id));
+
+    // Reset PointView to have new point table.
+    m_bufferPointView.reset(new PointView(*m_bufferPointTable));
+
+    if (m_info->dataType() == EptInfo::DataType::Laszip)
+        startId = readLaszip(*m_bufferPointView, key, m_nodeId);
+    else
+        startId = readBinary(*m_bufferPointView, key, m_nodeId);
+
+    m_currentIndex = 0;
+
+    // Read addon information after the native data, we'll possibly
+    // overwrite attributes.
+    for (const auto& addon : m_addons)
+        readAddon(*m_bufferPointView, key, *addon, startId);
+
+    m_nodeId++;
+}
+
+void EptReader::fillPoint(PointRef& point)
+{
+    DimTypeList dims = m_bufferPointView->dimTypes();
+    char* buffer = new char[m_bufferPointView->pointSize()];
+    m_bufferPointView->getPackedPoint(dims, m_currentIndex, buffer);
+    point.setPackedData(dims, buffer);
+    m_currentIndex++;
+    delete[] buffer;
+}
+
+bool EptReader::processOne(PointRef& point)
+{
+    // bufferView is not set if no tile has been read and
+    // currentIndex will be greater or equal to the view size when the
+    // whole tile has been read
+    bool finishedCurrentOverlap = !(m_bufferPointView &&
+        m_currentIndex < m_bufferPointView->size());
+
+    //We're done with all overlaps, Its time to finish reading.
+    if (m_nodeId > m_overlaps.size() && finishedCurrentOverlap)
+        return false;
+
+    //Either this is a first overlap or we've streamed all points
+    //from current overlap. So its time to load new overlap.
+    if (finishedCurrentOverlap)
+        loadNextOverlap();
+
+    fillPoint(point);
+
+    return true;
+}
+
 } // namespace pdal
 

io/EptReader.hpp

@@ -42,6 +42,7 @@
 
 #include <pdal/Reader.hpp>
 #include <pdal/util/Bounds.hpp>
+#include <pdal/Streamable.hpp>
 
 #include <nlohmann/json.hpp>
 
@@ -60,7 +61,7 @@ class FixedPointLayout;
 class Key;
 class Pool;
 
-class PDAL_DLL EptReader : public Reader
+class PDAL_DLL EptReader : public Reader, public Streamable
 {
     FRIEND_TEST(EptReaderTest, getRemoteType);
     FRIEND_TEST(EptReaderTest, getCoercedType);
@@ -102,6 +103,11 @@ class PDAL_DLL EptReader : public Reader
     static Dimension::Type getRemoteTypeTest(const NL::json& dimInfo);
     static Dimension::Type getCoercedTypeTest(const NL::json& dimInfo);
 
+    //For streamable pipeline.
+    virtual bool processOne(PointRef& point) override;
+    void loadNextOverlap();
+    void fillPoint(PointRef& point);
+
     std::string m_root;
 
     std::unique_ptr<arbiter::Arbiter> m_arbiter;
@@ -129,6 +135,13 @@ class PDAL_DLL EptReader : public Reader
 
     Dimension::Id m_nodeIdDim = Dimension::Id::Unknown;
     Dimension::Id m_pointIdDim = Dimension::Id::Unknown;
+
+    // For streamable pipeline.
+    uint64_t m_nodeId = 1;
+    std::unique_ptr<PointTable> m_bufferPointTable;
+    PointViewPtr m_bufferPointView;
+    PointLayoutPtr m_bufferLayout;
+    point_count_t m_currentIndex = -1;
 };
 
 } // namespace pdal

test/unit/io/EptReaderTest.cpp

@@ -467,4 +467,75 @@ TEST(EptReaderTest, getCoercedType)
     EXPECT_EQ(EptReader::getCoercedTypeTest(j), Dimension::Type::None);
 }
 
+void streamTest(const std::string src)
+{
+    Options ops;
+    ops.add("filename", src);
+    ops.add("resolution", 1);
+
+    EptReader eptReader;
+    eptReader.setOptions(ops);
+
+    PointTable t;
+    eptReader.prepare(t);
+    PointViewSet s = eptReader.execute(t);
+    PointViewPtr p = *s.begin();
+
+    class Checker : public Filter, public Streamable
+    {
+    public:
+        std::string getName() const
+        {
+            return "checker";
+        }
+        Checker(PointViewPtr v)
+            : m_cnt(0), m_view(v), m_bulkBuf(v->pointSize()),
+              m_buf(v->pointSize()), m_dims(v->dimTypes())
+        {
+        }
+
+    private:
+        point_count_t m_cnt;
+        PointViewPtr m_view;
+        std::vector<char> m_bulkBuf;
+        std::vector<char> m_buf;
+        DimTypeList m_dims;
+
+        bool processOne(PointRef& point)
+        {
+            PointRef bulkPoint = m_view->point(m_cnt);
+
+            bulkPoint.getPackedData(m_dims, m_bulkBuf.data());
+            point.getPackedData(m_dims, m_buf.data());
+            EXPECT_EQ(
+                memcmp(m_buf.data(), m_bulkBuf.data(), m_view->pointSize()), 0);
+            m_cnt++;
+            return true;
+        }
+
+        void done(PointTableRef)
+        {
+            EXPECT_EQ(m_cnt, m_view->size());
+        }
+    };
+
+    EptReader eptReader1;
+    eptReader1.setOptions(ops);
+
+    Checker c(p);
+    c.setInput(eptReader1);
+
+    FixedPointTable fixed(100);
+    c.prepare(fixed);
+    c.execute(fixed);
+}
+
+TEST(EptReaderTest, stream)
+{
+    streamTest(ellipsoidEptBinaryPath);
+#ifdef PDAL_HAVE_LASZIP
+    streamTest(eptLaszipPath);
+#endif
+}
+
 } // namespace pdal