feat!: align page_number metadata across splitters and fix drift in overlapping splits

docs-website/docs/pipeline-components/preprocessors/documentsplitter.mdx

@@ -33,7 +33,7 @@ description: "`DocumentSplitter` divides a list of text documents into a list of
 
 When `split_by="token"`, strings such as `<|endoftext|>` in the source document are tokenized as ordinary text. They are preserved in the chunks and count toward the token limit.
 
-A field `"source_id"` is added to each document's `meta` data to keep track of the original document that was split. Another meta field `"page_number"` is added to each document to keep track of the page it belonged to in the original document. Other metadata are copied from the original document.
+A field `"source_id"` is added to each document's `meta` data to keep track of the original document that was split. Another meta field `"page_number"` is added to each document with the page the chunk starts on, counted from the form feed (`\f`) characters in the original document. A chunk that spans a page break reports the page its text begins on. Other metadata are copied from the original document.
 
 The DocumentSplitter is compatible with the following DocumentStores:
 

docs-website/docs/pipeline-components/preprocessors/embeddingbaseddocumentsplitter.mdx

@@ -28,7 +28,7 @@ This component splits documents based on embedding similarity using cosine dista
 
 It first splits text into sentences, optionally groups them, calculates embeddings for each group, and then uses cosine
 distance between sequential embeddings to determine split points. Any distance above the specified percentile is treated
-as a break point. The component also tracks page numbers based on form feed characters (`\f`) in the original document.
+as a break point. The component also tracks page numbers based on form feed characters (`\f`) in the original document, reporting the page each chunk starts on.
 
 This component is inspired by [5 Levels of Text Splitting](https://github.com/FullStackRetrieval-com/RetrievalTutorials/blob/main/tutorials/LevelsOfTextSplitting/5_Levels_Of_Text_Splitting.ipynb) by Greg Kamradt.
 

docs-website/docs/pipeline-components/preprocessors/markdownheadersplitter.mdx

@@ -45,7 +45,7 @@ Parameters you can set when initializing the component:
 Each output document's metadata includes:
 
 - `source_id`: ID of the original document.
-- `page_number`: Page number. Updated when `page_break_character` is found.
+- `page_number`: The page the chunk starts on, counted from the `page_break_character` occurrences that precede its text.
 - `split_id`: Index of the chunk within its parent.
 - `header`: The header text for this chunk.
 - `parent_headers`: List of parent header texts in hierarchy order.

docs-website/docs/pipeline-components/preprocessors/recursivesplitter.mdx

@@ -33,6 +33,8 @@ The `RecursiveDocumentSplitter` expects a list of documents as input and returns
 
 When `split_unit="token"`, strings such as `<|endoftext|>` in the source document are tokenized as ordinary text. They are preserved in the chunks and count toward the token limit.
 
+A meta field `"page_number"` is added to each chunk with the page it starts on, counted from the form feed (`\f`) characters in the original document. A chunk that spans a page break reports the page its text begins on, matching [`DocumentSplitter`](documentsplitter.mdx).
+
 The separators are applied in the same order as they are defined in the list. The first separator is used on the text; any resulting chunk that is within the specified `chunk_size` is retained. For chunks that exceed the defined `chunk_size`, the next separator in the list is applied. If all separators are used and the chunk still exceeds the `chunk_size`, a hard split occurs based on the `chunk_size`, taking into account whether words or characters are used as counting units. This process is repeated until all chunks are within the limits of the specified `chunk_size`.
 
 ## Usage

haystack/components/preprocessors/_page_numbers.py

@@ -0,0 +1,38 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""
+The `page_number` convention shared by the splitters.
+
+Splitters record the page a chunk came from in its `page_number` metadata field, counting page breaks
+(form feed, "\f") in the source document. The page reported is the one the chunk's first non-page-break
+character lies on: a chunk that spans a page break belongs to the page it starts on, and page breaks a
+chunk opens with belong to that chunk's own page rather than to the page it was carved out of.
+"""
+
+
+def _leading_page_breaks(text: str, page_break_character: str = "\f") -> int:
+    """
+    Count how many pages a chunk's text starts past the page its first character is on.
+
+    This is the run of page breaks the chunk opens with, except for a chunk made up entirely of page
+    breaks: that chunk is an empty page and stays on the page it starts on, rather than being pushed onto
+    the following one. `split_by="page"` emits such chunks for every blank page.
+
+    :param text: The chunk to inspect.
+    :param page_break_character: The character sequence marking a page break.
+    :returns: The number of pages to advance, 0 if the chunk starts with anything but a page break or
+        consists only of page breaks.
+    """
+    if not page_break_character:
+        return 0
+
+    offset = 0
+    while text.startswith(page_break_character, offset):
+        offset += len(page_break_character)
+
+    # Nothing but page breaks: an empty page, which belongs to the page it starts on.
+    if offset >= len(text):
+        return 0
+    return offset // len(page_break_character)

haystack/components/preprocessors/document_splitter.py

@@ -9,6 +9,7 @@
 from more_itertools import windowed
 
 from haystack import Document, component, logging
+from haystack.components.preprocessors._page_numbers import _leading_page_breaks
 from haystack.components.preprocessors.sentence_tokenizer import Language, SentenceSplitter, nltk_imports
 from haystack.core.serialization import default_from_dict, default_to_dict
 from haystack.lazy_imports import LazyImport
@@ -199,7 +200,8 @@ def run(self, documents: list[Document]) -> dict[str, list[Document]]:
         :returns: A dictionary with the following key:
             - `documents`: List of documents with the split texts. Each document includes:
                 - A metadata field `source_id` to track the original document.
-                - A metadata field `page_number` to track the original page number.
+                - A metadata field `page_number` with the page the chunk starts on, counting form feed
+                  ("\f") characters in the original document.
                 - All other metadata copied from the original document.
 
         :raises TypeError: if the input is not a list of Documents.
@@ -450,7 +452,7 @@ def _create_docs_from_splits(
 
         for i, (txt, split_idx) in enumerate(zip(text_splits, splits_start_idxs, strict=True)):
             copied_meta = deepcopy(meta)
-            copied_meta["page_number"] = splits_pages[i]
+            copied_meta["page_number"] = splits_pages[i] + _leading_page_breaks(txt)
             copied_meta["split_id"] = i
             copied_meta["split_idx_start"] = split_idx
             doc = Document(content=txt, meta=copied_meta)

haystack/components/preprocessors/embedding_based_document_splitter.py

@@ -12,6 +12,7 @@
 
 from haystack import Document, component, logging
 from haystack.components.embedders.types import DocumentEmbedder
+from haystack.components.preprocessors._page_numbers import _leading_page_breaks
 from haystack.components.preprocessors.sentence_tokenizer import Language, SentenceSplitter
 from haystack.core.serialization import component_to_dict, default_from_dict, default_to_dict
 from haystack.utils.async_utils import _execute_component_async
@@ -179,7 +180,8 @@ def run(self, documents: list[Document]) -> dict[str, list[Document]]:
                 - A metadata field `source_id` to track the original document.
                 - A metadata field `split_id` to track the split number.
                 - A metadata field `split_idx_start` with the character offset of the chunk in the original document.
-                - A metadata field `page_number` to track the original page number.
+                - A metadata field `page_number` with the page the chunk starts on, counting form feed
+                  ("\f") characters in the original document.
                 - All other metadata copied from the original document.
 
         :raises RuntimeError: If the component wasn't warmed up.
@@ -220,7 +222,8 @@ async def run_async(self, documents: list[Document]) -> dict[str, list[Document]
                 - A metadata field `source_id` to track the original document.
                 - A metadata field `split_id` to track the split number.
                 - A metadata field `split_idx_start` with the character offset of the chunk in the original document.
-                - A metadata field `page_number` to track the original page number.
+                - A metadata field `page_number` with the page the chunk starts on, counting form feed
+                  ("\f") characters in the original document.
                 - All other metadata copied from the original document.
 
         :raises RuntimeError: If the component wasn't warmed up.
@@ -549,12 +552,11 @@ def _create_documents_from_splits(splits: list[str], original_doc: Document) ->
             split_meta["split_id"] = i
             split_meta["split_idx_start"] = current_char_pos
 
-            # Calculate page number for this split
-            # Count page breaks in the split itself
             page_breaks_in_split = split_text.count("\f")
 
-            # Calculate the page number for this split
-            split_meta["page_number"] = current_page
+            # current_page is the page this split's first character is on; a split that opens with page
+            # breaks starts its text on a later page.
+            split_meta["page_number"] = current_page + _leading_page_breaks(split_text)
 
             doc = Document(content=split_text, meta=split_meta)
             documents.append(doc)

haystack/components/preprocessors/markdown_header_splitter.py

@@ -8,6 +8,7 @@
 
 from haystack import Document, component, logging
 from haystack.components.preprocessors import DocumentSplitter
+from haystack.components.preprocessors._page_numbers import _leading_page_breaks
 
 logger = logging.getLogger(__name__)
 
@@ -294,8 +295,12 @@ def _apply_secondary_splitting(self, documents: list[tuple[Document, bool]]) ->
                     secondary_content_start_idx = header_match.end()
                     content_for_splitting = doc.content[secondary_content_start_idx:]
 
-            # The page this header chunk starts on; its splits are numbered relative to it.
-            chunk_start_page = doc.meta.get("page_number", 1)
+            # The page this header chunk's first character is on; its splits are numbered relative to it.
+            # doc.meta holds the page the chunk's *text* starts on, which already counts the breaks the chunk
+            # opens with, and page_break_ends below counts those same breaks again - so take them back off.
+            chunk_start_page = doc.meta.get("page_number", 1) - _leading_page_breaks(
+                doc.content, self.page_break_character
+            )
 
             clean_meta = {k: v for k, v in doc.meta.items() if k != "split_id"}
 
@@ -312,7 +317,11 @@ def _apply_secondary_splitting(self, documents: list[tuple[Document, bool]]) ->
                 while page_break_count < len(page_break_ends) and page_break_ends[page_break_count] <= split_start_idx:
                     page_break_count += 1
 
-                split.meta["page_number"] = chunk_start_page + page_break_count
+                split.meta["page_number"] = (
+                    chunk_start_page
+                    + page_break_count
+                    + _leading_page_breaks(split.content or "", self.page_break_character)
+                )
                 split.meta["split_id"] = current_split_id
                 if "source_id" in doc.meta:
                     split.meta["source_id"] = doc.meta["source_id"]
@@ -358,8 +367,11 @@ def _split_documents_by_markdown_headers(self, documents: list[Document]) -> lis
             )
             for split_idx, split in enumerate(splits):
                 meta = deepcopy(doc.meta) if doc.meta else {}
-                chunk_start_page = document_start_page + doc.content.count(
-                    self.page_break_character, 0, split["source_start_idx"]
+                chunk_start_page = (
+                    document_start_page
+                    + doc.content.count(self.page_break_character, 0, split["source_start_idx"])
+                    # breaks the chunk opens with end the previous page, so its text is on a later one
+                    + _leading_page_breaks(split["content"], self.page_break_character)
                 )
                 meta.update({"source_id": doc.id, "page_number": chunk_start_page, "split_id": split_idx})
                 from_header_split = split["from_header_split"]
@@ -386,7 +398,8 @@ def run(self, documents: list[Document]) -> dict[str, list[Document]]:
         :returns: A dictionary with the following key:
             - `documents`: List of documents with the split texts. Each document includes:
                 - A metadata field `source_id` to track the original document.
-                - A metadata field `page_number` to track the original page number.
+                - A metadata field `page_number` with the page the chunk starts on, counting
+                  `page_break_character` occurrences in the original document.
                 - A metadata field `split_id` to identify the split chunk index within its parent document.
                 - All other metadata copied from the original document.
         :raises ValueError: If a document has `None` content.