E18: content deduplication, exact and near

ccrawl/journal.go

@@ -122,6 +122,10 @@ type RunEvent struct {
 	// shard comes out smaller than expected.
 	LangDropped int64            `json:"lang_dropped,omitempty"`
 	LangCounts  map[string]int64 `json:"lang_counts,omitempty"`
+	// DigestDropped counts records --dedup-digest skipped as byte identical to
+	// one already emitted, for the same reason: a run that comes out smaller
+	// than expected should say which filter did it.
+	DigestDropped int64 `json:"digest_dropped,omitempty"`
 
 	Rate     float64 `json:"rate_per_hour,omitempty"`
 	ETAS     float64 `json:"eta_s,omitempty"`

ccrawl/markdown.go

@@ -23,15 +23,19 @@ import (
 // the open-index/open-markdown dataset layout so the output can be appended to
 // existing crawls without a schema migration.
 //
-// language, language_confidence, and extractor make this open-markdown-v3. They
-// are appended at the end and every earlier column keeps its name and type, so a
-// v2 reader projecting the columns it knows about reads a v3 file without
-// changes: parquet is read by name, and a reader that never asks for the new
-// columns never touches them.
+// language, language_confidence, extractor, and simhash make this
+// open-markdown-v3. They are appended at the end and every earlier column keeps
+// its name and type, so a v2 reader projecting the columns it knows about reads
+// a v3 file without changes: parquet is read by name, and a reader that never
+// asks for the new columns never touches them.
 //
 // extractor is name@version, not just the name. Extraction changes between
 // releases, sometimes visibly, and a dataset that only recorded the name could
 // not answer why two shards built months apart disagree about the same page.
+//
+// simhash is a fingerprint, not a decision. Collapsing near duplicates is a
+// judgement about what a corpus is for, and it belongs in a query the consumer
+// writes rather than in a pipeline that already threw the evidence away.
 type MarkdownRow struct {
 	DocID          string  `parquet:"doc_id"`
 	URL            string  `parquet:"url"`
@@ -44,6 +48,7 @@ type MarkdownRow struct {
 	Language       string  `parquet:"language"`
 	LangConfidence float64 `parquet:"language_confidence"`
 	Extractor      string  `parquet:"extractor"`
+	Simhash        uint64  `parquet:"simhash"`
 }
 
 // MarkdownDocID returns a stable 16-byte hex document ID derived from the URL.
@@ -77,6 +82,11 @@ type MarkdownStats struct {
 	// handed to Progress, so nothing reads the map while it is being written.
 	LangDropped int64
 	LangCounts  map[string]int64
+
+	// DigestDropped counts records --dedup-digest skipped because an identical
+	// payload had already been seen in this shard. It is filled in on the
+	// returned stats, after the reader goroutine has finished.
+	DigestDropped int64
 }
 
 // candidateRow is a converted document on its way to the writer, carrying the
@@ -128,6 +138,15 @@ type MarkdownPackConfig struct {
 	// Extractor is the engine that turns a captured page into Markdown. nil
 	// selects the default, which keeps every existing caller on h2m.
 	Extractor *Extractor
+	// DedupDigest skips a record whose payload is byte identical to one already
+	// seen in this shard. The check runs before conversion, so a dropped record
+	// costs a hash and not an extraction.
+	//
+	// The scope is one shard, not the whole run. Shards convert in parallel, so a
+	// run wide set would make which copy survives depend on which shard happened
+	// to finish first, and it would grow without a bound over --shards all. A
+	// per-shard set is deterministic and costs one entry per page.
+	DedupDigest bool
 	// Progress is called after each row is written. It may be nil.
 	Progress func(MarkdownStats)
 }
@@ -196,11 +215,14 @@ func packStream(ctx context.Context, body io.Reader, cfg MarkdownPackConfig, sta
 	rows := make(chan candidateRow, workers*4)
 	langCounts := map[string]int64{}
 
-	// Reader: iterate the source stream and push one record per page.
+	// Reader: iterate the source stream and push one record per page. Both
+	// results are read after the rows channel closes, which the reader's own
+	// close of records happens before, so neither needs a lock.
 	var readErr error
+	var digestDropped int64
 	go func() {
 		defer close(records)
-		readErr = streamSourceRecords(ctx, body, cfg, ex, records)
+		digestDropped, readErr = streamSourceRecords(ctx, body, cfg, ex, records)
 	}()
 
 	tConvert := time.Now()
@@ -234,6 +256,7 @@ func packStream(ctx context.Context, body io.Reader, cfg MarkdownPackConfig, sta
 						Language:       code,
 						LangConfidence: conf,
 						Extractor:      exID,
+						Simhash:        Simhash(md),
 					},
 				}
 			}
@@ -268,6 +291,7 @@ func packStream(ctx context.Context, body io.Reader, cfg MarkdownPackConfig, sta
 		}
 	}
 	stats.LangCounts = langCounts
+	stats.DigestDropped = digestDropped
 
 	stats.DurConvert = time.Since(tConvert)
 
@@ -301,16 +325,35 @@ func newMarkdownParquetWriter(path string) (*ParquetWriter[MarkdownRow], error)
 	return &ParquetWriter[MarkdownRow]{f: f, w: w}, nil
 }
 
-// streamSourceRecords reads the shard and pushes one record per page. A WARC
-// shard carries HTML in response records; a WET shard carries text Common Crawl
-// already extracted, in conversion records. Both come out as the same struct,
-// so everything downstream of here is identical for the two sources and only
-// the extractor knows the difference.
-func streamSourceRecords(ctx context.Context, body io.Reader, cfg MarkdownPackConfig, ex *Extractor, records chan<- htmlRecord) error {
+// streamSourceRecords reads the shard and pushes one record per page, returning
+// how many records --dedup-digest skipped. A WARC shard carries HTML in response
+// records; a WET shard carries text Common Crawl already extracted, in
+// conversion records. Both come out as the same struct, so everything downstream
+// of here is identical for the two sources and only the extractor knows the
+// difference.
+func streamSourceRecords(ctx context.Context, body io.Reader, cfg MarkdownPackConfig, ex *Extractor, records chan<- htmlRecord) (int64, error) {
 	queued := 0
+	var dropped int64
 	errStop := errors.New("record limit reached")
 
+	// Exact duplicates are the cheap half of deduplication and the bigger half by
+	// volume: the same page served on two URLs, a mirror, a session id in the
+	// query string. Hashing the payload before conversion means a duplicate costs
+	// a hash instead of an extraction.
+	var seen map[[sha256.Size]byte]struct{}
+	if cfg.DedupDigest {
+		seen = make(map[[sha256.Size]byte]struct{}, 8192)
+	}
+
 	push := func(rec htmlRecord) error {
+		if seen != nil {
+			sum := sha256.Sum256(rec.html)
+			if _, dup := seen[sum]; dup {
+				dropped++
+				return nil
+			}
+			seen[sum] = struct{}{}
+		}
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
@@ -354,9 +397,9 @@ func streamSourceRecords(ctx context.Context, body io.Reader, cfg MarkdownPackCo
 		})
 	}
 	if errors.Is(err, errStop) {
-		return nil // a deliberate early stop is not a read failure
+		return dropped, nil // a deliberate early stop is not a read failure
 	}
-	return err
+	return dropped, err
 }
 
 // convertGated runs the extractor while holding a slot in sem, so the total

ccrawl/markdown_run.go

@@ -119,6 +119,8 @@ type MarkdownExportConfig struct {
 	// Extractor is the engine every shard in the run uses. nil selects the
 	// default.
 	Extractor *Extractor
+	// DedupDigest skips pages whose bytes were already seen in the same shard.
+	DedupDigest bool
 
 	// Progress is called once per committed batch with a snapshot of the run.
 	// It may be nil.
@@ -159,6 +161,9 @@ type MarkdownRunStats struct {
 	// LangCounts[lang].
 	LangDropped int64
 	LangCounts  map[string]int64
+	// DigestDropped is how many records --dedup-digest skipped across every
+	// shard in the run.
+	DigestDropped int64
 }
 
 // packShardFn is the function the orchestrator uses to convert one shard. It
@@ -284,6 +289,7 @@ func RunMarkdownExport(ctx context.Context, h *HTTPClient, hf *HFClient, cfg Mar
 					Lang:              cfg.Lang,
 					MinLangConfidence: cfg.MinLangConfidence,
 					Extractor:         cfg.Extractor,
+					DedupDigest:       cfg.DedupDigest,
 				})
 				inflight.Add(-1)
 				// Per-shard wall-clock is the useful convert figure for a parallel
@@ -349,26 +355,27 @@ func markdownTickEvent(kind string, s *MarkdownRunStats, cfg MarkdownExportConfi
 		eta = float64(remaining) / rate * 3600
 	}
 	return RunEvent{
-		Event:        kind,
-		Crawl:        cfg.CrawlID,
-		Done:         s.Committed + s.Skipped + s.Failed,
-		Total:        s.Total,
-		Committed:    s.Committed,
-		Skipped:      s.Skipped,
-		Failed:       s.Failed,
-		Inflight:     inflight,
-		Rows:         s.Rows,
-		WARCBytes:    s.WARCBytes,
-		HTMLBytes:    s.HTMLBytes,
-		MDBytes:      s.MDBytes,
-		ParquetBytes: s.ParquetBytes,
-		LangDropped:  s.LangDropped,
-		LangCounts:   s.LangCounts,
-		Rate:         rate,
-		ETAS:         eta,
-		ElapsedS:     elapsed.Seconds(),
-		FreeDisk:     freeDiskBytes(cfg.OutDir),
-		RSS:          currentRSSBytes(),
+		Event:         kind,
+		Crawl:         cfg.CrawlID,
+		Done:          s.Committed + s.Skipped + s.Failed,
+		Total:         s.Total,
+		Committed:     s.Committed,
+		Skipped:       s.Skipped,
+		Failed:        s.Failed,
+		Inflight:      inflight,
+		Rows:          s.Rows,
+		WARCBytes:     s.WARCBytes,
+		HTMLBytes:     s.HTMLBytes,
+		MDBytes:       s.MDBytes,
+		ParquetBytes:  s.ParquetBytes,
+		LangDropped:   s.LangDropped,
+		LangCounts:    s.LangCounts,
+		DigestDropped: s.DigestDropped,
+		Rate:          rate,
+		ETAS:          eta,
+		ElapsedS:      elapsed.Seconds(),
+		FreeDisk:      freeDiskBytes(cfg.OutDir),
+		RSS:           currentRSSBytes(),
 	}
 }
 
@@ -409,6 +416,7 @@ func runCommitter(ctx context.Context, hf *HFClient, cfg MarkdownExportConfig, k
 			run.ParquetBytes += r.stats.ParquetBytes
 			run.ConvertS += int64(r.stats.DurConvert.Seconds())
 			run.LangDropped += r.stats.LangDropped
+			run.DigestDropped += r.stats.DigestDropped
 			for code, n := range r.stats.LangCounts {
 				if run.LangCounts == nil {
 					run.LangCounts = map[string]int64{}
@@ -470,18 +478,19 @@ func runCommitter(ctx context.Context, hf *HFClient, cfg MarkdownExportConfig, k
 		for _, r := range batch {
 			idx := r.idx
 			cfg.Reporter.Event(RunEvent{
-				Event:        EventShard,
-				Crawl:        cfg.CrawlID,
-				Shard:        &idx,
-				Status:       StatusOK,
-				Rows:         r.stats.Rows,
-				WARCBytes:    r.stats.WARCBytes,
-				HTMLBytes:    r.stats.HTMLBytes,
-				MDBytes:      r.stats.MDBytes,
-				ParquetBytes: r.stats.ParquetBytes,
-				LangDropped:  r.stats.LangDropped,
-				LangCounts:   r.stats.LangCounts,
-				ConvertS:     r.stats.DurConvert.Seconds(),
+				Event:         EventShard,
+				Crawl:         cfg.CrawlID,
+				Shard:         &idx,
+				Status:        StatusOK,
+				Rows:          r.stats.Rows,
+				WARCBytes:     r.stats.WARCBytes,
+				HTMLBytes:     r.stats.HTMLBytes,
+				MDBytes:       r.stats.MDBytes,
+				ParquetBytes:  r.stats.ParquetBytes,
+				LangDropped:   r.stats.LangDropped,
+				LangCounts:    r.stats.LangCounts,
+				DigestDropped: r.stats.DigestDropped,
+				ConvertS:      r.stats.DurConvert.Seconds(),
 			})
 		}
 		batch = batch[:0]

ccrawl/refetch.go

@@ -50,6 +50,7 @@ type RefetchRow struct {
 	Language       string  `parquet:"language"`
 	LangConfidence float64 `parquet:"language_confidence"`
 	Extractor      string  `parquet:"extractor"`
+	Simhash        uint64  `parquet:"simhash"`
 }
 
 // RefetchStats summarises one shard's refetch run with per-phase breakdown.
@@ -89,6 +90,10 @@ type RefetchStats struct {
 	LangDropped int64
 	LangCounts  map[string]int64
 
+	// DigestDropped is how many rows --dedup-digest kept out of the parquet
+	// because an identical response body had already been written.
+	DigestDropped int64
+
 	// Phase 4: parquet write.
 	ParquetBytes int64
 	DurExport    time.Duration
@@ -136,6 +141,13 @@ type RefetchPackConfig struct {
 	// returns HTML, and there is no Common Crawl text to pass through.
 	Extractor *Extractor
 
+	// DedupDigest keeps only the first row for each response body digest in this
+	// shard. The fetch has already happened by then, so unlike the export
+	// pipeline this saves storage rather than work. Which copy of a duplicate
+	// survives depends on which fetch finished first; how many were dropped does
+	// not.
+	DedupDigest bool
+
 	// CacheDir, when set, is where the downloaded WARC is cached so a re-run of
 	// the same shard skips the multi-second download. The download streams to a
 	// .part file beside the final name and is renamed into place only once it is
@@ -241,6 +253,14 @@ func PackRefetchShard(ctx context.Context, h *HTTPClient, cfg RefetchPackConfig)
 		ex = Extractors[DefaultExtractor]
 	}
 	exID := ex.ID(cfg.CrawlID)
+	// The digest is the one ami already computed over the response body, so
+	// deduplication here costs a map lookup. Failed fetches are exempt: they have
+	// no body to be a duplicate of, and dropping all but one of them would hide
+	// the dead hosts a refetch run exists to find.
+	var seenDigest map[string]struct{}
+	if cfg.DedupDigest {
+		seenDigest = make(map[string]struct{}, 8192)
+	}
 
 	var wg sync.WaitGroup
 	wg.Add(convertWorkers)
@@ -289,6 +309,7 @@ func PackRefetchShard(ctx context.Context, h *HTTPClient, cfg RefetchPackConfig)
 						row.MarkdownLength = int64(len(md))
 						row.Markdown = md
 						row.Language, row.LangConfidence = DetectLanguage(md)
+						row.Simhash = Simhash(md)
 					}
 				}
 				rows <- row
@@ -312,18 +333,27 @@ func PackRefetchShard(ctx context.Context, h *HTTPClient, cfg RefetchPackConfig)
 			if row.FinalURL != "" && row.FinalURL != row.URL {
 				stats.Redirected++
 			}
-			if row.Markdown != "" || row.HTML != "" {
-				stats.Rows++
-			}
 		}
 		// The stats above count the whole attempt, failures included, because
-		// that is what tells you how the crawl went. The filter decides only what
-		// reaches the parquet.
+		// that is what tells you how the crawl went. The filters below decide what
+		// reaches the parquet, and Rows is counted after them: it is the number
+		// the dataset card publishes, so it has to match the file rather than the
+		// attempt.
+		if seenDigest != nil && row.Error == "" && row.Digest != "" {
+			if _, dup := seenDigest[row.Digest]; dup {
+				stats.DigestDropped++
+				continue
+			}
+			seenDigest[row.Digest] = struct{}{}
+		}
 		langCounts[row.Language]++
 		if !LangMatches(cfg.Lang, row.Language, row.LangConfidence, minConf) {
 			stats.LangDropped++
 			continue
 		}
+		if row.Markdown != "" || row.HTML != "" {
+			stats.Rows++
+		}
 		if werr := pw.Write(row); werr != nil {
 			go func() {
 				for range rows {