- Document table: Added parent_id and document_type columns with index

- Document schema (knowledge/document.py): Added document_id field for
  streaming retrieval
- Librarian operations:
  - add-child-document for extracted PDF pages
  - list-children to get child documents
  - stream-document for chunked content retrieval
  - Cascade delete removes children when parent is deleted
  - list-documents filters children by default
- PDF decoder (decoding/pdf/pdf_decoder.py): Updated to stream large
  documents from librarian API to temp file
- Librarian service (librarian/service.py): Sends document_id instead of
  content for large PDFs (>2MB)
- Deprecated tools (load_pdf.py, load_text.py): Added deprecation
  warnings directing users to tg-add-library-document +
  tg-start-library-processing

trustgraph-flow/trustgraph/decoding/pdf/pdf_decoder.py

@@ -2,8 +2,12 @@
 """
 Simple decoder, accepts PDF documents on input, outputs pages from the
 PDF document as text as separate output objects.
+
+Supports both inline document data and streaming from librarian API
+for large documents.
 """
 
+import os
 import tempfile
 import base64
 import logging
@@ -17,6 +21,10 @@ logger = logging.getLogger(__name__)
 
 default_ident = "pdf-decoder"
 
+# Default API URL for fetching documents from librarian
+default_api_url = os.environ.get("TRUSTGRAPH_API_URL", "http://api:8088")
+
+
 class Processor(FlowProcessor):
 
     def __init__(self, **params):
@@ -29,6 +37,8 @@ class Processor(FlowProcessor):
             }
         )
 
+        self.api_url = params.get("api_url", default_api_url)
+
         self.register_specification(
             ConsumerSpec(
                 name = "input",
@@ -46,6 +56,59 @@ class Processor(FlowProcessor):
 
         logger.info("PDF decoder initialized")
 
+    def _fetch_document_to_file(self, document_id, user, file_path, chunk_size=1024*1024):
+        """
+        Fetch document content from librarian API and stream to file.
+
+        This avoids loading the entire document into memory at once.
+        """
+        import requests
+
+        logger.info(f"Streaming document {document_id} to temp file...")
+
+        # Use chunk-based streaming to minimize memory usage
+        chunk_index = 0
+        total_bytes = 0
+
+        with open(file_path, 'wb') as f:
+            while True:
+                url = f"{self.api_url}/api/v1/librarian"
+                payload = {
+                    "operation": "stream-document",
+                    "user": user,
+                    "document-id": document_id,
+                    "chunk-index": chunk_index,
+                    "chunk-size": chunk_size,
+                }
+
+                try:
+                    response = requests.post(url, json=payload, timeout=60)
+                    response.raise_for_status()
+                    data = response.json()
+                except requests.RequestException as e:
+                    logger.error(f"Failed to fetch chunk {chunk_index}: {e}")
+                    raise
+
+                if "error" in data and data["error"]:
+                    raise RuntimeError(f"API error: {data['error']}")
+
+                content_b64 = data.get("content", "")
+                if not content_b64:
+                    break
+
+                chunk_data = base64.b64decode(content_b64)
+                f.write(chunk_data)
+                total_bytes += len(chunk_data)
+
+                total_chunks = data.get("total-chunks", 1)
+                if chunk_index >= total_chunks - 1:
+                    break
+
+                chunk_index += 1
+
+        logger.info(f"Downloaded {total_bytes} bytes to temp file")
+        return total_bytes
+
     async def on_message(self, msg, consumer, flow):
 
         logger.debug("PDF message received")
@@ -54,26 +117,43 @@ class Processor(FlowProcessor):
 
         logger.info(f"Decoding PDF {v.metadata.id}...")
 
-        with tempfile.NamedTemporaryFile(delete_on_close=False) as fp:
+        with tempfile.NamedTemporaryFile(delete_on_close=False, suffix='.pdf') as fp:
+            temp_path = fp.name
 
-            fp.write(base64.b64decode(v.data))
-            fp.close()
+            # Check if we should fetch from librarian or use inline data
+            if v.document_id:
+                # Stream from librarian API to temp file
+                logger.info(f"Fetching document {v.document_id} from librarian...")
+                fp.close()
+                self._fetch_document_to_file(
+                    document_id=v.document_id,
+                    user=v.metadata.user,
+                    file_path=temp_path,
+                )
+            else:
+                # Use inline data (backward compatibility)
+                fp.write(base64.b64decode(v.data))
+                fp.close()
 
-            with open(fp.name, mode='rb') as f:
+            loader = PyPDFLoader(temp_path)
+            pages = loader.load()
 
-                loader = PyPDFLoader(fp.name)
-                pages = loader.load()
+            for ix, page in enumerate(pages):
 
-                for ix, page in enumerate(pages):
+                logger.debug(f"Processing page {ix}")
 
-                    logger.debug(f"Processing page {ix}")
+                r = TextDocument(
+                    metadata=v.metadata,
+                    text=page.page_content.encode("utf-8"),
+                )
 
-                    r = TextDocument(
-                        metadata=v.metadata,
-                        text=page.page_content.encode("utf-8"),
-                    )
+                await flow("output").send(r)
 
-                    await flow("output").send(r)
+        # Clean up temp file
+        try:
+            os.unlink(temp_path)
+        except OSError:
+            pass
 
         logger.debug("PDF decoding complete")
 
@@ -81,6 +161,12 @@ class Processor(FlowProcessor):
     def add_args(parser):
         FlowProcessor.add_args(parser)
 
+        parser.add_argument(
+            '--api-url',
+            default=default_api_url,
+            help=f'TrustGraph API URL for document streaming (default: {default_api_url})',
+        )
+
 def run():
 
     Processor.launch(default_ident, __doc__)

trustgraph-flow/trustgraph/librarian/librarian.py

@@ -91,6 +91,21 @@ class Librarian:
         ):
             raise RuntimeError("Document does not exist")
 
+        # First, cascade delete all child documents
+        children = await self.table_store.list_children(request.document_id)
+        for child in children:
+            logger.debug(f"Cascade deleting child document {child.id}")
+            try:
+                child_object_id = await self.table_store.get_document_object_id(
+                    child.user,
+                    child.id
+                )
+                await self.blob_store.remove(child_object_id)
+                await self.table_store.remove_document(child.user, child.id)
+            except Exception as e:
+                logger.warning(f"Failed to delete child document {child.id}: {e}")
+
+        # Now remove the parent document
         object_id = await self.table_store.get_document_object_id(
             request.user,
             request.document_id
@@ -262,6 +277,14 @@ class Librarian:
 
         docs = await self.table_store.list_documents(request.user)
 
+        # Filter out child documents by default unless include_children is True
+        include_children = getattr(request, 'include_children', False)
+        if not include_children:
+            docs = [
+                doc for doc in docs
+                if not doc.parent_id  # Only include top-level documents
+            ]
+
         return LibrarianResponse(
             error = None,
             document_metadata = None,
@@ -593,3 +616,124 @@ class Librarian:
             upload_sessions=upload_sessions,
         )
 
+    # Child document operations
+
+    async def add_child_document(self, request):
+        """
+        Add a child document linked to a parent document.
+
+        Child documents are typically extracted content (e.g., pages from a PDF).
+        They have a parent_id pointing to the source document and document_type
+        set to "extracted".
+        """
+        logger.info(f"Adding child document {request.document_metadata.id} "
+                   f"for parent {request.document_metadata.parent_id}")
+
+        if not request.document_metadata.parent_id:
+            raise RequestError("parent_id is required for child documents")
+
+        # Verify parent exists
+        if not await self.table_store.document_exists(
+                request.document_metadata.user,
+                request.document_metadata.parent_id
+        ):
+            raise RequestError(
+                f"Parent document {request.document_metadata.parent_id} does not exist"
+            )
+
+        if await self.table_store.document_exists(
+                request.document_metadata.user,
+                request.document_metadata.id
+        ):
+            raise RequestError("Document already exists")
+
+        # Ensure document_type is set to "extracted"
+        request.document_metadata.document_type = "extracted"
+
+        # Create object ID for blob
+        object_id = uuid.uuid4()
+
+        logger.debug("Adding blob...")
+
+        await self.blob_store.add(
+            object_id, base64.b64decode(request.content),
+            request.document_metadata.kind
+        )
+
+        logger.debug("Adding to table...")
+
+        await self.table_store.add_document(
+            request.document_metadata, object_id
+        )
+
+        logger.debug("Add child document complete")
+
+        return LibrarianResponse(
+            error=None,
+            document_id=request.document_metadata.id,
+        )
+
+    async def list_children(self, request):
+        """
+        List all child documents for a given parent document.
+        """
+        logger.debug(f"Listing children for parent {request.document_id}")
+
+        children = await self.table_store.list_children(request.document_id)
+
+        return LibrarianResponse(
+            error=None,
+            document_metadatas=children,
+        )
+
+    async def stream_document(self, request):
+        """
+        Stream document content in chunks.
+
+        This operation returns document content in smaller chunks, allowing
+        memory-efficient processing of large documents. The response includes
+        chunk information for reassembly.
+
+        Note: This operation returns a single chunk at a time. Clients should
+        call repeatedly with increasing chunk_index until all chunks are received.
+        """
+        logger.debug(f"Streaming document {request.document_id}, chunk {request.chunk_index}")
+
+        object_id = await self.table_store.get_document_object_id(
+            request.user,
+            request.document_id
+        )
+
+        # Default chunk size of 1MB
+        chunk_size = request.chunk_size if request.chunk_size > 0 else 1024 * 1024
+
+        # Get the full content and slice out the requested chunk
+        # Note: This is a simple implementation. For true streaming, we'd need
+        # range requests on the object storage.
+        content = await self.blob_store.get(object_id)
+        total_size = len(content)
+        total_chunks = math.ceil(total_size / chunk_size)
+
+        if request.chunk_index >= total_chunks:
+            raise RequestError(
+                f"Invalid chunk index {request.chunk_index}, "
+                f"document has {total_chunks} chunks"
+            )
+
+        start = request.chunk_index * chunk_size
+        end = min(start + chunk_size, total_size)
+        chunk_content = content[start:end]
+
+        logger.debug(f"Returning chunk {request.chunk_index}/{total_chunks}, "
+                    f"bytes {start}-{end} of {total_size}")
+
+        return LibrarianResponse(
+            error=None,
+            content=base64.b64encode(chunk_content),
+            chunk_index=request.chunk_index,
+            chunks_received=1,  # Using as "current chunk" indicator
+            total_chunks=total_chunks,
+            bytes_received=end,
+            total_bytes=total_size,
+        )
+

trustgraph-flow/trustgraph/librarian/service.py

@@ -271,6 +271,9 @@ class Processor(AsyncProcessor):
 
         pass
 
+    # Threshold for sending document_id instead of inline content (2MB)
+    STREAMING_THRESHOLD = 2 * 1024 * 1024
+
     async def load_document(self, document, processing, content):
 
         logger.debug("Ready for document processing...")
@@ -303,15 +306,31 @@ class Processor(AsyncProcessor):
             )
             schema = TextDocument
         else:
-            doc = Document(
-                metadata = Metadata(
-                    id = document.id,
-                    metadata = document.metadata,
-                    user = processing.user,
-                    collection = processing.collection
-                ),
-                data = base64.b64encode(content).decode("utf-8")
-            )
+            # For large PDF documents, send document_id for streaming retrieval
+            # instead of embedding the entire content in the message
+            if len(content) >= self.STREAMING_THRESHOLD:
+                logger.info(f"Document {document.id} is large ({len(content)} bytes), "
+                           f"sending document_id for streaming retrieval")
+                doc = Document(
+                    metadata = Metadata(
+                        id = document.id,
+                        metadata = document.metadata,
+                        user = processing.user,
+                        collection = processing.collection
+                    ),
+                    document_id = document.id,
+                    data = b"",  # Empty data, receiver will fetch via API
+                )
+            else:
+                doc = Document(
+                    metadata = Metadata(
+                        id = document.id,
+                        metadata = document.metadata,
+                        user = processing.user,
+                        collection = processing.collection
+                    ),
+                    data = base64.b64encode(content).decode("utf-8")
+                )
             schema = Document
 
         logger.debug(f"Submitting to queue {q}...")
@@ -368,6 +387,10 @@ class Processor(AsyncProcessor):
             "abort-upload": self.librarian.abort_upload,
             "get-upload-status": self.librarian.get_upload_status,
             "list-uploads": self.librarian.list_uploads,
+            # Child document and streaming operations
+            "add-child-document": self.librarian.add_child_document,
+            "list-children": self.librarian.list_children,
+            "stream-document": self.librarian.stream_document,
         }
 
         if v.operation not in impls:

trustgraph-flow/trustgraph/tables/library.py

@@ -112,6 +112,34 @@ class LibraryTableStore:
             ON document (object_id)
         """);
 
+        # Add parent_id and document_type columns for child document support
+        logger.debug("document table parent_id column...")
+
+        try:
+            self.cassandra.execute("""
+                ALTER TABLE document ADD parent_id text
+            """);
+        except Exception as e:
+            # Column may already exist
+            if "already exists" not in str(e).lower() and "Invalid column name" not in str(e):
+                logger.debug(f"parent_id column may already exist: {e}")
+
+        try:
+            self.cassandra.execute("""
+                ALTER TABLE document ADD document_type text
+            """);
+        except Exception as e:
+            # Column may already exist
+            if "already exists" not in str(e).lower() and "Invalid column name" not in str(e):
+                logger.debug(f"document_type column may already exist: {e}")
+
+        logger.debug("document parent index...")
+
+        self.cassandra.execute("""
+            CREATE INDEX IF NOT EXISTS document_parent
+            ON document (parent_id)
+        """);
+
         logger.debug("processing table...")
 
         self.cassandra.execute("""
@@ -162,9 +190,10 @@ class LibraryTableStore:
             (
                 id, user, time,
                 kind, title, comments,
-                metadata, tags, object_id
+                metadata, tags, object_id,
+                parent_id, document_type
             )
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         """)
 
         self.update_document_stmt = self.cassandra.prepare("""
@@ -175,7 +204,8 @@ class LibraryTableStore:
         """)
 
         self.get_document_stmt = self.cassandra.prepare("""
-            SELECT time, kind, title, comments, metadata, tags, object_id
+            SELECT time, kind, title, comments, metadata, tags, object_id,
+                   parent_id, document_type
             FROM document
             WHERE user = ? AND id = ?
         """)
@@ -194,14 +224,16 @@ class LibraryTableStore:
 
         self.list_document_stmt = self.cassandra.prepare("""
             SELECT
-                id, time, kind, title, comments, metadata, tags, object_id
+                id, time, kind, title, comments, metadata, tags, object_id,
+                parent_id, document_type
             FROM document
             WHERE user = ?
         """)
 
         self.list_document_by_tag_stmt = self.cassandra.prepare("""
             SELECT
-                id, time, kind, title, comments, metadata, tags, object_id
+                id, time, kind, title, comments, metadata, tags, object_id,
+                parent_id, document_type
             FROM document
             WHERE user = ? AND tags CONTAINS ?
             ALLOW FILTERING
@@ -277,6 +309,16 @@ class LibraryTableStore:
             WHERE user = ?
         """)
 
+        # Child document queries
+        self.list_children_stmt = self.cassandra.prepare("""
+            SELECT
+                id, user, time, kind, title, comments, metadata, tags,
+                object_id, parent_id, document_type
+            FROM document
+            WHERE parent_id = ?
+            ALLOW FILTERING
+        """)
+
     async def document_exists(self, user, id):
 
         resp = self.cassandra.execute(
@@ -303,6 +345,10 @@ class LibraryTableStore:
             for v in document.metadata
         ]
 
+        # Get parent_id and document_type from document, defaulting if not set
+        parent_id = getattr(document, 'parent_id', '') or ''
+        document_type = getattr(document, 'document_type', 'source') or 'source'
+
         while True:
 
             try:
@@ -312,7 +358,8 @@ class LibraryTableStore:
                     (
                         document.id, document.user, int(document.time * 1000),
                         document.kind, document.title, document.comments,
-                        metadata, document.tags, object_id
+                        metadata, document.tags, object_id,
+                        parent_id, document_type
                     )
                 )
 
@@ -419,6 +466,55 @@ class LibraryTableStore:
                     for m in row[5]
                 ],
                 tags = row[6] if row[6] else [],
+                parent_id = row[8] if row[8] else "",
+                document_type = row[9] if row[9] else "source",
+            )
+            for row in resp
+        ]
+
+        logger.debug("Done")
+
+        return lst
+
+    async def list_children(self, parent_id):
+        """List all child documents for a given parent document ID."""
+
+        logger.debug(f"List children for parent {parent_id}")
+
+        while True:
+
+            try:
+
+                resp = self.cassandra.execute(
+                    self.list_children_stmt,
+                    (parent_id,)
+                )
+
+                break
+
+            except Exception as e:
+                logger.error("Exception occurred", exc_info=True)
+                raise e
+
+        lst = [
+            DocumentMetadata(
+                id = row[0],
+                user = row[1],
+                time = int(time.mktime(row[2].timetuple())),
+                kind = row[3],
+                title = row[4],
+                comments = row[5],
+                metadata = [
+                    Triple(
+                        s=tuple_to_term(m[0], m[1]),
+                        p=tuple_to_term(m[2], m[3]),
+                        o=tuple_to_term(m[4], m[5])
+                    )
+                    for m in row[6]
+                ],
+                tags = row[7] if row[7] else [],
+                parent_id = row[9] if row[9] else "",
+                document_type = row[10] if row[10] else "source",
             )
             for row in resp
         ]
@@ -464,6 +560,8 @@ class LibraryTableStore:
                     for m in row[4]
                 ],
                 tags = row[5] if row[5] else [],
+                parent_id = row[7] if row[7] else "",
+                document_type = row[8] if row[8] else "source",
             )
 
             logger.debug("Done")