diff --git a/dbms/src/Processors/QueryPipeline.cpp b/dbms/src/Processors/QueryPipeline.cpp
index 6fbc3bb8ff1..06fed2ac3fc 100644
--- a/dbms/src/Processors/QueryPipeline.cpp
+++ b/dbms/src/Processors/QueryPipeline.cpp
@@ -515,8 +515,8 @@ void QueryPipeline::setProgressCallback(const ProgressCallback & callback)
 {
     for (auto & processor : processors)
     {
-        if (auto * source = typeid_cast<SourceFromInputStream *>(processor.get()))
-            source->getStream().setProgressCallback(callback);
+        if (auto * source = typeid_cast<ISourceWithProgress *>(processor.get()))
+            source->setProgressCallback(callback);
 
         if (auto * source = typeid_cast<CreatingSetsTransform *>(processor.get()))
             source->setProgressCallback(callback);
@@ -527,8 +527,8 @@ void QueryPipeline::setProcessListElement(QueryStatus * elem)
 {
     for (auto & processor : processors)
     {
-        if (auto * source = typeid_cast<SourceFromInputStream *>(processor.get()))
-            source->getStream().setProcessListElement(elem);
+        if (auto * source = dynamic_cast<ISourceWithProgress *>(processor.get()))
+            source->setProcessListElement(elem);
 
         if (auto * source = typeid_cast<CreatingSetsTransform *>(processor.get()))
             source->setProcessListElement(elem);