From bc1b38c99879ab9a0f3ed2556a438f0ee1973ea4 Mon Sep 17 00:00:00 2001
From: cybermaggedon <cybermaggedon@gmail.com>
Date: Thu, 31 Oct 2024 10:10:53 +0000
Subject: [PATCH] Fix schema match (#131)

---
 .../trustgraph/dump/triples/parquet/processor.py          | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/trustgraph-parquet/trustgraph/dump/triples/parquet/processor.py b/trustgraph-parquet/trustgraph/dump/triples/parquet/processor.py
index 553a62b8..dc15d8a9 100755
--- a/trustgraph-parquet/trustgraph/dump/triples/parquet/processor.py
+++ b/trustgraph-parquet/trustgraph/dump/triples/parquet/processor.py
@@ -9,7 +9,7 @@
 import argparse
 import time
 
-from .... schema import Triple
+from .... schema import Triples
 from .... schema import triples_store_queue
 from .... base import Consumer
 
@@ -38,7 +38,7 @@ def __init__(self, **params):
             **params | {
                 "input_queue": input_queue,
                 "subscriber": subscriber,
-                "input_schema": Triple,
+                "input_schema": Triples,
             }
         )
 
@@ -51,7 +51,9 @@ def __del__(self):
     def handle(self, msg):
 
         v = msg.value()
-        self.writer.write(v.s.value, v.p.value, v.o.value)
+
+        for t in v.triples:
+            self.writer.write(t.s.value, t.p.value, t.o.value)
 
     @staticmethod
     def add_args(parser):