Document Ingestion: Clean Text, Metadata and Incremental Updates from PDF, HTML and Markdown

Version note: The code in this post was built and run with .NET 10, PdfPig 0.1.16, AngleSharp 1.10.0, Microsoft.Extensions.DataIngestion 10.11.0-preview.1, CommunityToolkit.VectorData.PgVector 1.0.2 and Npgsql 10.0.3 against PostgreSQL 17.11 + pgvector 0.8.7 and Ollama 0.34.4 (nomic-embed-text).

Guess First

Bookstore's corporate sales agreement is a two-page PDF (Portable Document Format). It contains the sentence "liste fiyatı üzerinden %15 indirim" (a 15% discount on the list price). You extract the text with a library and search it with text.Contains("fiyat") ("fiyat" means price). Is the result true?

It is not as silly a question as it looks. In this post we look at the least discussed and most error-prone part of the RAG (Retrieval-Augmented Generation) pipeline: reading documents, cleaning them, carrying them into chunks with their metadata, and keeping the index up to date as documents change. Both the "stale data" and "chunk boundary" failures from the first post are born here.

Every Format Comes with Its Own Noise

Our document folder has three formats: Markdown policy files, an HTML (HyperText Markup Language) page exported from the help center, and a PDF agreement. We want to turn all of them into the same structure, a list of sections (Section) with a heading and a body, so the chunker does not need to know the format.

Markdown is the easiest: headings are already marked and metadata is read from the block at the top of the file. In HTML, the real text gets lost between the menu, the cookie banner and the footer. If we take the page's body text as is with AngleSharp:

$ dotnet run -- html-naive
Ana Sayfa | Kampanyalar | Yardım | Hesabım | Sepetim (0) Bu site çerez kullanır. Devam ederek çerez politikasını kabul etmiş olursun. Tamam Sıkça Sorulan Sorular Siparişimi nasıl iptal ederim? Siparişin kargoya verilmeden önce Hesabım > Siparişlerim sayfasından iptal edebilirsin. Kargoya verilmiş siparişler iptal edilemez; teslim aldıktan sonra iade talebi oluşturman gerekir. Fatura adresimi değiştirebilir miyim? Fat ...

The chunk's first sentence is "Ana Sayfa | Kampanyalar | Yardım | Hesabım | Sepetim (0)", the site menu. Because this menu and the cookie text are the same on every page, once they reach the embedding they make all help pages look alike and show up as answers to unrelated questions such as "cookie policy". Cleaning means recognizing the noise by its structure and dropping it:

public static SourceDocument LoadHtml(string path, string rel)
{
    var document = new HtmlParser().ParseDocument(File.ReadAllText(path));
    string Meta(string name) =>
        document.QuerySelector($"meta[name='bookstore:{name}']")?.GetAttribute("content") ?? "";

    // Noise: menu, cookie banner, footer and scripts must not reach the embedding
    foreach (var noise in document.QuerySelectorAll("script, style, nav, header, footer, .cookie").ToList())
    {
        noise.Remove();
    }

    List<Section> sections = document.QuerySelectorAll("main section")
        .Select(s => new Section(
            s.QuerySelector("h2")!.TextContent.Trim(),
            string.Join("\n\n", s.QuerySelectorAll("p").Select(p => p.TextContent.Trim()))))
        .ToList();
    string title = document.QuerySelector("main h1")!.TextContent.Trim();
    return new(Meta("id"), title, Meta("tenant"), Meta("acl"),
        DateOnly.Parse(Meta("updated_at"), CultureInfo.InvariantCulture), rel, sections);
}

Metadata (id, acl, updated_at) comes from the page's meta tags. In a real help center it is sturdier to get this from the CMS's (Content Management System) API (Application Programming Interface); scraping the page is the last resort.

Clean Text from PDF

PDF is not a text format but a drawing format: it places specific glyphs at specific coordinates on a page. There is no notion of "paragraph", "heading" or "table"; you have to recover them from the coordinates. Let's look at PdfPig's page text as is:

$ dotnet run -- pdf-naive
[page 1] Kurumsal Satış SözleşmesiBu sözleşme, okul, kütüphane, şirket ve kamu kurumlarının (bundan sonra "Kurum") Bookstore'dantoplu kitap alımına ilişkin koşulları düzenler. Yürürlük tarihi: 1 Eylül 2026.1. Toplu alım indirimiAynı siparişte aynı kitaptan en az 50 adet alan Kurum, liste fiyatı üzerinden %15 indirim alır. 200adet ...

Three problems at once. Line breaks are gone and words are glued together: "SözleşmesiBu", "Bookstore'dantoplu", "200adet". The table has turned into one meaningless string (later in the output: "Adetİndirim50-199%15200 ve üzeri%22"). The third is invisible: the "fi" in "liste fiyatı" is not two letters but a single Unicode character (U+FB01, the "fi" ligature). The answer to the opening question:

$ dotnet run -- ligature
Contains("fiyat"): False
NFKC Contains("fiyat"): True
o200k tokens: 4 vs 3

Contains("fiyat") returned false. Text with a ligature is never found by keyword search, and on the embedding side it is split into different tokens. Unicode's NFKC (Normalization Form Compatibility Composition) normalization maps such compatibility characters to their equivalents. Turkish letters (ı, İ, ş, ğ) are not affected by it; still, do not apply normalization without testing it on your own text.

The second step is rebuilding lines: join the words on the same baseline from left to right and sort the lines from top to bottom. Then headers and footers: lines repeated on every page. My first attempt dropped lines that were exactly identical:

$ dotnet run -- pdf-v1
...
Kurumsal siparişler, stok durumuna bağlı olarak sipariş onayından sonra 10 iş günü içinde
Kurum'un bildirdiği tek adrese teslim edilir. Teslimat ücreti 100 adet ve üzeri siparişlerde
Bookstore'a, daha küçük siparişlerde Kurum'a aittir.
Sayfa 1 / 2
Birden fazla adrese dağıtım isteniyorsa her ek adres için ayrı teslimat ücreti alınır. Teslimatta eksik
veya hasarlı ürün, teslim tutanağına işlenmeli ve 3 iş günü içinde yazılı olarak bildirilmelidir.
...

The header and the fixed footer line are gone, but "Sayfa 1 / 2" (page 1 of 2) stayed and landed in the middle of the "3. Teslimat" (delivery) section, because on the second page the same line is "Sayfa 2 / 2". The fix is to mask digits before comparing. The final version:

List<List<string>> pages = [];
using (PdfDocument pdf = PdfDocument.Open(path))
{
    foreach (var page in pdf.GetPages())
    {
        // Group words into lines: same baseline (Y) = same line
        List<string> pageLines = page.GetWords()
            .GroupBy(w => Math.Round(w.BoundingBox.Bottom))
            .OrderByDescending(g => g.Key)
            .Select(g => string.Join(' ', g.OrderBy(w => w.BoundingBox.Left).Select(w => w.Text)))
            .Select(line => line.Normalize(NormalizationForm.FormKC)) // "fi" -> "fi", "fl" -> "fl"
            .ToList();
        pages.Add(pageLines);
    }
}

// A line repeated on every page = header/footer; mask digits so "Sayfa 1 / 2" is caught too
static string Shape(string line) => Digits().Replace(line, "#");
HashSet<string> repeated = pages.Count > 1
    ? pages.Skip(1).Aggregate(pages[0].Select(Shape).ToHashSet(),
        (set, p) => { set.IntersectWith(p.Select(Shape)); return set; })
    : [];

Sections are split at numbered headings such as "1. Toplu alım indirimi", and a section that crosses a page boundary stays one section. The table is still flat text ("Adet İndirim 50-199 %15 200 ve üzeri %22"); that is one reason the model said 22% for 120 copies in the first post. To keep tables with their structure you either get the data from the system that produced the PDF or use a table-aware parser; a plain text extractor does not do it.

Metadata on Every Chunk

Ingestion produces more than text and a vector. In the series' doc_chunks table every chunk carries the following, and each has a reason:

• doc_id and source_path: the way back to the document for citations, deletes and reprocessing.
• title and section: the heading path; it is also prepended to the chunk text, because a sentence like "for orders paid with a gift card..." does not say on its own which policy it belongs to.
• tenant and acl: the retrieval filter; the cure for the staff discount leak in the first post.
• updated_at: a freshness filter and the answer to "how current is this?"
• content_hash: the hash of the whole document, so an unchanged document is not embedded again.
• embedding_model: which model produced the vector; protection against the model-switch trap from the second post.

A page number is also valuable for PDFs (to be able to say "agreement, page 2" in a citation); I did not keep it because the section number is enough for our two-page agreement. The rule: write into the chunk at ingestion time everything you will want to filter on during retrieval or show in the answer. Adding it later means reprocessing every document.

The Same Pipeline with Microsoft.Extensions.DataIngestion

.NET's own ingestion library, Microsoft.Extensions.DataIngestion, is still in preview (this post uses 10.11.0-preview.1). It has four building blocks: a reader that turns a document into an IngestionDocument (Markdig and MarkItDown packages), a chunker that splits it (HeaderChunker, SectionChunker, SemanticSimilarityChunker, DocumentTokenChunker), processors that enrich chunks, and VectorStoreWriter<T>, which writes the result to a vector store. IngestionPipeline<T> ties them together:

using Microsoft.Extensions.AI;
using Microsoft.Extensions.DataIngestion;
using Microsoft.Extensions.Logging;
using Microsoft.ML.Tokenizers;
using CommunityToolkit.VectorData.PgVector;
using OllamaSharp;

string connectionString = $"Host=localhost;Port=19701;Database=ragmedi;Username=postgres;Password={Environment.GetEnvironmentVariable("PGPW")}";
IEmbeddingGenerator<string, Embedding<float>> embedder =
    new OllamaApiClient(new Uri("http://localhost:11434"), "nomic-embed-text");

// Reader: turns Markdown into an IngestionDocument (sections, headers, paragraphs, tables)
IngestionDocumentReader reader = new MarkdownReader();

// Chunker: split by headers, at most 300 tokens per chunk
Tokenizer tokenizer = TiktokenTokenizer.CreateForEncoding("o200k_base");
IngestionChunker<string> chunker = new HeaderChunker(new IngestionChunkerOptions(tokenizer)
{
    MaxTokensPerChunk = 300,
    OverlapTokens = 0
});

// Writer: embeds the chunks and writes them to the "medi_chunks" collection in pgvector
using PostgresVectorStore store = new(connectionString, new PostgresVectorStoreOptions { EmbeddingGenerator = embedder });
using VectorStoreWriter<string> writer = new(store, dimensionCount: 768, new VectorStoreWriterOptions
{
    CollectionName = "medi_chunks",
    IncrementalIngestion = true
});

using ILoggerFactory loggerFactory = LoggerFactory.Create(b => b.AddSimpleConsole(o => o.SingleLine = true).SetMinimumLevel(LogLevel.Warning));
using IngestionPipeline<string> pipeline = new(reader, chunker, writer, loggerFactory: loggerFactory);

await foreach (IngestionResult result in pipeline.ProcessAsync(new DirectoryInfo(args[0]), "*.md", SearchOption.AllDirectories))
{
    string status = result.Succeeded ? "ok" : $"FAILED {result.Exception?.GetType().Name}: {result.Exception?.Message}";
    Console.WriteLine($"{result.DocumentId,-60} {status}");
}

I ran this program on a copy of the policy and product files. I deliberately added an unreadable file to the folder (kampanya.md, with its permissions removed):

$ dotnet run -- demo/medi
demo/medi/products/urun-katalogu.md ok
demo/medi/policies/hediye-karti.md ok
demo/medi/policies/kampanya.md FAILED UnauthorizedAccessException: Access to the path 'demo/medi/policies/kampanya.md' is denied.
fail: Microsoft.Extensions.DataIngestion.IngestionPipeline[6] An error occurred while ingesting document 'demo/medi/policies/kampanya.md'. ...
demo/medi/policies/odeme-secenekleri.md ok
demo/medi/policies/kargo-ve-teslimat.md ok
demo/medi/policies/iade-politikasi.md ok
demo/medi/policies/sadakat-programi.md ok

The unreadable file did not stop the whole run: ProcessAsync returns an IngestionResult for every document, the failed one has Succeeded false and its Exception set, and the pipeline's logger also wrote a separate fail line. 36 chunks were written from the other six files. HeaderChunker puts the heading path both into the chunk's Context and at the start of its text ("# Hediye Kartı ## Geçerlilik süresi"), the same thing our chunker does. Let's look at the table the writer created:

\d medi_chunks
                     Table "public.medi_chunks"
   Column   |    Type     | Collation | Nullable |      Default      
------------+-------------+-----------+----------+-------------------
 key        | uuid        |           | not null | gen_random_uuid()
 content    | text        |           |          | 
 context    | text        |           |          | 
 documentid | text        |           |          | 
 embedding  | vector(768) |           |          | 
Indexes:
    "medi_chunks_pkey" PRIMARY KEY, btree (key)
    "medi_chunks_documentid_index" btree (documentid)

The default schema has key, content, context, documentid and embedding; no acl, tenant or updated_at. The metadata block at the top of the Markdown files was not carried into any chunk (zero chunks contain acl:), and documentid is the file's full path. So as it is, no permission filter can be built; for metadata you need to write a chunk processor or use a writer with your own record type. IncrementalIngestion = true deletes a document's old chunks when it is reprocessed. Let's see what happens to deleted files: I removed the product catalog from the folder and ran the pipeline again:

$ mv demo/medi/products demo/products_removed
$ dotnet run -- demo/medi
demo/medi/policies/hediye-karti.md ok
demo/medi/policies/odeme-secenekleri.md ok
demo/medi/policies/kargo-ve-teslimat.md ok
demo/medi/policies/iade-politikasi.md ok
demo/medi/policies/sadakat-programi.md ok

SELECT document, count(*) FROM medi_chunks GROUP BY 1;
           document            | count 
-------------------------------+-------
 policies/hediye-karti.md      |     3
 policies/iade-politikasi.md   |     5
 policies/kargo-ve-teslimat.md |     6
 policies/odeme-secenekleri.md |     4
 policies/sadakat-programi.md  |     4
 products/urun-katalogu.md     |    14
(6 rows)

The policy files were reprocessed and their chunk counts stayed the same, so there are no duplicates; the old chunks really were replaced. But although urun-katalogu.md is no longer in the folder, its 14 chunks are still in the index. The pipeline only knows the files it is given; deleting the chunks of a file that no longer exists is your job.

Incremental Updates: Hash, Upsert and Delete

Ingestion does not run once and finish. The shipping policy changes, new products arrive, an old agreement is withdrawn. Re-embedding everything every time is no problem for a small folder, but for hundreds of thousands of chunks it means hours and needless cost. Incremental updates have three rules: skip what did not change, replace the old chunks of a changed document with new ones in a single transaction, and delete from the index what is no longer in the source.

public sealed record SyncReport(int Added, int Updated, int Unchanged, int Removed, int EmbeddedChunks, long ElapsedMs);

public sealed class IncrementalIndexer(NpgsqlDataSource dataSource, IEmbeddingGenerator<string, Embedding<float>> embedder)
{
    public static string DocumentHash(SourceDocument doc) =>
        Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(
            string.Join("\n", doc.Sections.Select(s => s.Heading + "\n" + s.Body)))))[..16];

    public async Task<SyncReport> SyncAsync(IReadOnlyList<SourceDocument> source, CancellationToken ct = default)
    {
        Stopwatch sw = Stopwatch.StartNew();
        await using NpgsqlConnection conn = await dataSource.OpenConnectionAsync(ct);

        // Which document is in the index with which hash?
        Dictionary<string, string> indexed = [];
        await using (NpgsqlCommand cmd = new("SELECT DISTINCT doc_id, content_hash FROM doc_chunks", conn))
        await using (NpgsqlDataReader r = await cmd.ExecuteReaderAsync(ct))
        {
            while (await r.ReadAsync(ct)) indexed[r.GetString(0)] = r.GetString(1);
        }

        int added = 0, updated = 0, unchanged = 0, embedded = 0;
        foreach (SourceDocument doc in source)
        {
            string hash = DocumentHash(doc);
            if (indexed.TryGetValue(doc.Id, out string? old) && old == hash)
            {
                unchanged++;
                continue;
            }

            // Changed document: chunk, embed, delete old chunks and write new ones in one transaction
            List<Chunk> chunks = Chunkers.Section([doc]);
            GeneratedEmbeddings<Embedding<float>> vectors =
                await embedder.GenerateAsync(chunks.Select(c => "search_document: " + c.Content), cancellationToken: ct);
            await using NpgsqlTransaction tx = await conn.BeginTransactionAsync(ct);
            await using (NpgsqlCommand del = new("DELETE FROM doc_chunks WHERE doc_id = @id", conn, tx))
            {
                del.Parameters.AddWithValue("id", doc.Id);
                await del.ExecuteNonQueryAsync(ct);
            }
            for (int i = 0; i < chunks.Count; i++)
            {
                Chunk c = chunks[i];
                await using NpgsqlCommand ins = new("""
                    INSERT INTO doc_chunks (id, doc_id, title, section, content, source_path, tenant, acl,
                                            updated_at, content_hash, embedding_model, embedding)
                    VALUES (@id, @doc, @title, @section, @content, @path, @tenant, @acl, @updated, @hash, @model, @embedding)
                    """, conn, tx);
                ins.Parameters.AddWithValue("id", c.Id);
                ins.Parameters.AddWithValue("doc", c.DocId);
                ins.Parameters.AddWithValue("title", c.Title);
                ins.Parameters.AddWithValue("section", c.Section);
                ins.Parameters.AddWithValue("content", c.Content);
                ins.Parameters.AddWithValue("path", c.SourcePath);
                ins.Parameters.AddWithValue("tenant", c.Tenant);
                ins.Parameters.AddWithValue("acl", c.Acl);
                ins.Parameters.AddWithValue("updated", c.UpdatedAt);
                ins.Parameters.AddWithValue("hash", hash);
                ins.Parameters.AddWithValue("model", vectors[i].ModelId ?? "unknown");
                ins.Parameters.AddWithValue("embedding", new Vector(vectors[i].Vector));
                await ins.ExecuteNonQueryAsync(ct);
            }
            await tx.CommitAsync(ct);
            embedded += chunks.Count;
            if (old is null) added++; else updated++;
        }

        // Delete the vectors of documents that are no longer in the source
        string[] gone = indexed.Keys.Except(source.Select(d => d.Id)).ToArray();
        await using (NpgsqlCommand del = new("DELETE FROM doc_chunks WHERE doc_id = ANY(@ids)", conn))
        {
            del.Parameters.AddWithValue("ids", gone);
            await del.ExecuteNonQueryAsync(ct);
        }

        return new(added, updated, unchanged, gone.Length, embedded, sw.ElapsedMilliseconds);
    }
}

The document hash is computed from the sections' headings and bodies, that is, from the extracted and cleaned text. Hashing the file bytes is also possible, but then changing the parser or a cleaning rule does not reprocess unchanged documents; a hash of the text catches that too. Delete and insert run in the same transaction: a query that arrives in between sees either the old chunks or the new ones, never half a document. I ran the scenario in four steps:

$ dotnet run -- sync demo/sync --archive
SyncReport { Added = 11, Updated = 0, Unchanged = 0, Removed = 0, EmbeddedChunks = 55, ElapsedMs = 257 }

$ dotnet run -- sync demo/sync
SyncReport { Added = 0, Updated = 0, Unchanged = 10, Removed = 1, EmbeddedChunks = 0, ElapsedMs = 38 }

$ dotnet run -- sync demo/sync
SyncReport { Added = 0, Updated = 0, Unchanged = 10, Removed = 0, EmbeddedChunks = 0, ElapsedMs = 36 }

$ sed -i "" "s/400 TL/450 TL/g" demo/sync/policies/kargo-ve-teslimat.md
$ dotnet run -- sync demo/sync
SyncReport { Added = 0, Updated = 1, Unchanged = 9, Removed = 0, EmbeddedChunks = 6, ElapsedMs = 176 }

The first run imitates the state of an old index: the archive folder is part of the source and 55 chunks of 11 documents were written. In the second run the archive folder is no longer a source: 10 documents were skipped because they did not change, zero embeddings were produced, and the old 2025 shipping policy was deleted from the index; the source of the "same-day delivery" answer from the first post is gone. The third run did nothing: idempotent. In the fourth I raised the free shipping threshold in the shipping policy from 400 to 450 TL; only that document was updated and only its 6 chunks were embedded again. Doing the same change with a "reload everything" approach would have embedded 53 chunks. Do not read the ElapsedMs values as a speed measurement: in these experiments embeddings produced earlier came from a disk cache in the work folder; only the 6 new chunks in the fourth step went to Ollama.

What runs this sync is a background job: periodic (every night or every hour) or triggered by a change event from the document system. Which one to choose and how to run the job reliably is the subject of the Background Jobs series; from RAG's point of view what matters is that the job is idempotent. Running the same sync twice must do nothing the second time; the hash check ensures that.

Production Notes

Every ingestion run should produce a report: how many documents were added, changed, skipped, deleted, how many failed and how many chunks were embedded. If one night you see "deleted: 412", you notice that someone moved the folder before your customers do. An upper bound on deletions (for example, never delete more than 20% of the index in one run without approval) is cheap insurance.

Design for partial success. One corrupt file should not stop the whole ingestion, but the failed file should not silently disappear either. Log failures with the file name and retry them on the next run.

Version your parser and cleaning rules. When you change a rule such as the PDF header cleanup, every affected document has to be reprocessed. A hash of the text does this automatically; if you use a byte hash, add a "pipeline version" field.

Keep the original document. Chunks are a derivative; one day, when you change the chunk size, the embedding model or the metadata schema, you must be able to regenerate everything from the source.

Common Pitfalls

First, trusting text that comes out of a PDF. Ligatures, glued words, headers and footers in the middle of pages, flattened tables. Write "expected text" tests for a few sample PDFs in ingestion; if a parser update breaks them, you see it immediately.

Second, binding tightly to a preview API. Microsoft.Extensions.DataIngestion is promising but in preview; names and behavior can change between versions. If you keep a thin adapter between it and your own Section and Chunk types, you fix a single file when the library changes.

Third, trying to add metadata later. In the first post the acl field was in the document, but since it was not carried into the chunks no permission filter could be built. Write metadata with the chunk from the first day of ingestion.

Fourth, forgetting deleted documents. An ingestion that only processes existing files leaves the vectors of a deleted document in the index forever; the "same-day delivery" answer in the first post was born exactly like that.

Fifth, letting noise into chunks. Menus, cookie banners, page numbers and lines repeated on every page such as "Gizli değildir" (not confidential) make embeddings look alike and lower recall.

A Small Experiment

★☆☆ Remove the Normalize(NormalizationForm.FormKC) line from LoadPdf and, for the question "Kurumsal siparişlerde fiyat bilgisi gizli mi?" (is price information confidential for corporate orders?), check the rank of the "5. Gizlilik" section with the recall setup from the second post. Does the ligature affect retrieval?

★★☆ Add a safety rule to IncrementalIndexer: if a single sync would delete more than 20% of the documents in the index, throw an exception without deleting. Try it by replacing the source folder with an empty one.

★★★ Add page numbers to PDF sections: put a Page field on Section, record the page where the section heading appears and add a page column to doc_chunks. The "3. Teslimat" section spans two pages, so decide which page to store and justify your decision.

Now we have clean text, metadata and an index that stays current. But how big should the pieces be? In the next post we compare fixed-size, header-based, semantic and parent-child chunking with the same question set and try Anthropic's Contextual Retrieval method.