diff --git a/src/main/java/com/teragrep/functions/dpf_02/BatchCollect.java b/src/main/java/com/teragrep/functions/dpf_02/BatchCollect.java
index 1af66d6..4368803 100644
--- a/src/main/java/com/teragrep/functions/dpf_02/BatchCollect.java
+++ b/src/main/java/com/teragrep/functions/dpf_02/BatchCollect.java
@@ -132,13 +132,14 @@ public void collect(Dataset<Row> batchDF, Long batchId, List<AbstractStep> postB
         if (!skipLimiting && limit > 0) {
             orderedDs = orderedDs.limit(limit);
         }
-
+        List<Row> collected = orderedDs.collectAsList();
+        Dataset<Row> createdDsFromCollected = SparkSession.builder().getOrCreate().createDataFrame(collected, this.inputSchema);
         Dataset<Row> current;
         if (this.savedDs == null) {
-            current = orderedDs;
+            current = createdDsFromCollected;
         }
         else {
-            current = savedDs.union(orderedDs);
+            current = savedDs.union(createdDsFromCollected);
         }
 
         current = orderDataset(current);