OSM

vespa/repository/api/ingestion/processor.py

@@ -292,9 +292,17 @@ async def process_batch(batch):
     count = 0
     filters = dataset_config.get('files')[transformer_index].get('filters')
 
+
+
+
+    async for document in stream:
+        while count < limit:
+            logger.info(document)
+            continue
+
+
+
     async for document in stream:
-        logger.info(document)
-        continue
 
         # Apply filters (if any)
         if filters and not any(f(document) for f in filters):
@@ -308,10 +316,6 @@ async def process_batch(batch):
         # Process the batch when it reaches the batch_size or limit
         if len(current_batch) >= batch_size or (limit is not None and count >= limit):
 
-            # Temporarily skip processing TODO: REMOVE
-            logger.info(document)
-            continue
-
             batch_results = await process_batch(current_batch)
             results.extend(batch_results)  # Collect results
             current_batch = []