feat(doc): add position-aware PDF extraction with table detection

Phase 2: Improve PDF text extraction with lopdf for position-aware
text ordering and table detection.

Changes:
- Enhance parse_pdf_with_lopdf to preserve row structure and detect
  potential table rows based on consistent column alignment
- Add [_TABLE_ROW_] markers for downstream table detection
- Add process_table_row_markers to parse marked table rows into
  structured table blocks with extraction method "lopdf-position-aware"
- Add detect_table_row for position-based table row detection
- Add PositionedTextItem struct for tracking text positions
- Improve row grouping with y-coordinate scaling for tolerance

Author: RoyLin

core/src/composite_document_parser/extract/text.rs

@@ -90,6 +90,28 @@ pub(super) fn fallback_text_blocks(text: &str) -> Vec<DocumentBlock> {
 }
 
 pub(super) fn paged_text_blocks(text: &str, default_kind: DocumentBlockKind) -> Vec<DocumentBlock> {
+    // First, handle page breaks from lopdf extraction
+    let pages = text
+        .split("[_PAGE_BREAK_]")
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .collect::<Vec<_>>();
+
+    if pages.len() > 1 {
+        // Multi-page document: process each page separately
+        return pages
+            .into_iter()
+            .flat_map(|page| paged_text_blocks(page, default_kind.clone()))
+            .collect();
+    }
+
+    // Single page or after page split: handle table row markers
+    let blocks = process_table_row_markers(text, default_kind.clone());
+    if !blocks.is_empty() {
+        return blocks;
+    }
+
+    // Fall back to regular chunk processing
     split_paged_chunks(text)
         .into_iter()
         .flat_map(|chunk| {
@@ -106,6 +128,74 @@ pub(super) fn paged_text_blocks(text: &str, default_kind: DocumentBlockKind) ->
         .collect()
 }
 
+/// Process text with [_TABLE_ROW_] markers from lopdf position-aware extraction.
+///
+/// Groups consecutive table rows into a single table block.
+/// Returns empty Vec if no [_TABLE_ROW_] markers were found, allowing
+/// normal multi-column layout detection to proceed.
+fn process_table_row_markers(text: &str, default_kind: DocumentBlockKind) -> Vec<DocumentBlock> {
+    // Fast path: if no table row markers exist, return empty to use normal processing
+    if !text.contains("[_TABLE_ROW_]") {
+        return Vec::new();
+    }
+
+    let mut result = Vec::new();
+    let mut current_text = String::new();
+    let mut table_rows: Vec<String> = Vec::new();
+
+    for line in text.lines() {
+        let trimmed = line.trim();
+        if trimmed.starts_with("[_TABLE_ROW_]") && trimmed.ends_with("[_TABLE_ROW_]") {
+            // Extract the actual row content
+            let row_content = trimmed
+                .strip_prefix("[_TABLE_ROW_]")
+                .and_then(|s| s.strip_suffix("[_TABLE_ROW_]"))
+                .unwrap_or(trimmed)
+                .trim();
+
+            if !row_content.is_empty() {
+                table_rows.push(row_content.to_string());
+            }
+        } else if !table_rows.is_empty() {
+            // End of table rows - flush the accumulated table
+            if let Some(table_block) = build_table_from_rows(&table_rows) {
+                result.push(table_block);
+            }
+            table_rows.clear();
+            // Also flush any pending text
+            if !current_text.trim().is_empty() {
+                result.extend(chunk_to_blocks(&current_text.trim(), default_kind.clone()));
+                current_text.clear();
+            }
+            // Process this non-table line
+            if !trimmed.is_empty() {
+                current_text.push_str(trimmed);
+                current_text.push('\n');
+            }
+        } else {
+            // Normal text, accumulate
+            if !trimmed.is_empty() {
+                current_text.push_str(trimmed);
+                current_text.push('\n');
+            }
+        }
+    }
+
+    // Flush remaining table rows
+    if !table_rows.is_empty() {
+        if let Some(table_block) = build_table_from_rows(&table_rows) {
+            result.push(table_block);
+        }
+    }
+
+    // Flush remaining text
+    if !current_text.trim().is_empty() {
+        result.extend(chunk_to_blocks(&current_text.trim(), default_kind));
+    }
+
+    result
+}
+
 pub(super) fn text_blocks(text: &str, default_kind: DocumentBlockKind) -> Vec<DocumentBlock> {
     let normalized = super::normalize_text(text);
     normalized
@@ -783,3 +873,56 @@ pub(super) fn split_aligned_columns_with_gaps(line: &str) -> AlignedTextRow {
         }
     }
 }
+
+/// Build a table DocumentBlock from parsed table rows.
+fn build_table_from_rows(rows: &[String]) -> Option<DocumentBlock> {
+    if rows.len() < 2 {
+        return None;
+    }
+
+    // Try to parse each row as tab or comma-separated
+    let mut table_rows: Vec<Vec<String>> = Vec::new();
+    for row in rows {
+        let cells: Vec<String> = if row.contains('\t') {
+            row.split('\t')
+                .map(str::trim)
+                .filter(|s| !s.is_empty())
+                .map(String::from)
+                .collect()
+        } else {
+            // Try comma separation
+            row.split(',')
+                .map(str::trim)
+                .filter(|s| !s.is_empty())
+                .map(String::from)
+                .collect()
+        };
+
+        if cells.len() >= 2 {
+            table_rows.push(cells);
+        }
+    }
+
+    if table_rows.len() < 2 {
+        return None;
+    }
+
+    // Verify consistent column count
+    let col_count = table_rows[0].len();
+    if !table_rows.iter().all(|r| r.len() == col_count) {
+        return None;
+    }
+
+    let row_count = table_rows.len();
+    let column_count = col_count;
+    let content = super::table_text_from_cells(&table_rows);
+    let payload = super::table_structured_payload(&table_rows)?;
+
+    Some(
+        DocumentBlock::new(DocumentBlockKind::Table, Some("pdf-table"), content)
+            .with_attribute("row_count", row_count.to_string())
+            .with_attribute("column_count", column_count.to_string())
+            .with_attribute("extraction", "lopdf-position-aware")
+            .with_structured_payload(payload),
+    )
+}

core/src/composite_document_parser/ocr/mod.rs

@@ -374,6 +374,17 @@ fn parse_pdf(path: &Path) -> Result<String> {
         .with_context(|| format!("failed to extract text from PDF {}", path.display()))
 }
 
+/// Text item with position for table detection.
+#[derive(Debug, Clone)]
+struct PositionedTextItem {
+    page: usize,
+    y: f32,
+    x: f32,
+    text: String,
+    /// Y coordinate scaled to integer for grouping
+    y_scaled: i32,
+}
+
 /// Extract text from PDF using lopdf for better position-aware extraction.
 ///
 /// This provides improved text ordering (top-to-bottom, left-to-right) compared to
@@ -390,10 +401,8 @@ fn parse_pdf_with_lopdf(path: &Path) -> Result<String> {
         anyhow::bail!("PDF has no pages: {}", path.display());
     }
 
-    // Track text with positions: (page, y_coord_inverted, x_coord, text)
-    // Using BTreeMap for automatic sorting
-    // Scale coordinates to integers to work around f32 not implementing Ord
-    let mut all_text: BTreeMap<(usize, i32, i32), String> = BTreeMap::new();
+    // Collect all text items with positions
+    let mut all_items: Vec<PositionedTextItem> = Vec::new();
 
     // Collect all pages and sort by page number
     let mut page_list: Vec<(u32, lopdf::ObjectId)> = pages.iter().map(|(&k, &v)| (k, v)).collect();
@@ -406,38 +415,135 @@ fn parse_pdf_with_lopdf(path: &Path) -> Result<String> {
             let text_items = extract_text_from_content_stream(&contents, page_num as usize);
             for (y, x, text) in text_items {
                 if !text.trim().is_empty() {
-                    // Invert y so higher values come first (top of page)
-                    // Scale by 1000 to preserve decimal precision as integers
-                    let y_key = -(y * 1000.0) as i32;
-                    let x_key = (x * 100.0) as i32;
-                    all_text.insert((page_num as usize, y_key, x_key), text);
+                    // Scale y by 1000 to group rows (tolerance of ~1 pixel for 72dpi)
+                    let y_scaled = (y * 1000.0) as i32;
+                    all_items.push(PositionedTextItem {
+                        page: page_num as usize,
+                        y,
+                        x,
+                        text,
+                        y_scaled,
+                    });
                 }
             }
         }
     }
 
-    // Build output with page markers
+    if all_items.is_empty() {
+        anyhow::bail!("PDF has no extractable text: {}", path.display());
+    }
+
+    // Build output with page markers and row structure preserved
     let mut output = String::new();
     let mut current_page = 0usize;
 
-    for ((page_num, _, _), text) in all_text {
+    // Group items by page and y position (row grouping)
+    let mut page_groups: BTreeMap<usize, BTreeMap<i32, Vec<PositionedTextItem>>> = BTreeMap::new();
+    for item in all_items {
+        page_groups
+            .entry(item.page)
+            .or_default()
+            .entry(item.y_scaled)
+            .or_default()
+            .push(item);
+    }
+
+    for (page_num, y_groups) in page_groups {
         // Add page break marker when page changes
         if page_num != current_page {
             if current_page > 0 {
                 output.push_str("\n\n[_PAGE_BREAK_]\n\n");
             }
             current_page = page_num;
         }
-        // Preserve position hints for alignment detection
-        if !output.is_empty() && !output.ends_with('\n') {
-            output.push(' ');
+
+        // Process rows in order (BTreeMap is sorted by y_scaled)
+        for (_y_key, mut items) in y_groups {
+            // Sort items within row by x position
+            items.sort_by(|a, b| {
+                let a_x = (a.x * 100.0) as i32;
+                let b_x = (b.x * 100.0) as i32;
+                let x_cmp = a_x.cmp(&b_x);
+                x_cmp.then_with(|| a.text.cmp(&b.text))
+            });
+
+            // Join items on same row
+            let row_text = items
+                .iter()
+                .map(|item| item.text.as_str())
+                .collect::<Vec<_>>()
+                .join(" ");
+
+            // Detect potential table rows by checking:
+            // 1. Row has multiple cell-like fragments (separated by tabs or consistent spacing)
+            // 2. Row looks like it could be a table row (multiple aligned segments)
+            let is_potential_table_row = detect_table_row(&items);
+
+            if is_potential_table_row {
+                // Mark as potential table row for downstream table detection
+                output.push_str(&format!("[_TABLE_ROW_]{}[_TABLE_ROW_]", row_text));
+            } else {
+                output.push_str(&row_text);
+            }
+            output.push('\n');
         }
-        output.push_str(&text);
     }
 
     Ok(output)
 }
 
+/// Detect if a row of text items might be part of a table.
+///
+/// Looks for patterns like:
+/// - Multiple tab-separated cells
+/// - Multiple fragments with consistent horizontal spacing (column alignment)
+/// - Fragments that look like table cells (short text, consistent width)
+fn detect_table_row(items: &[PositionedTextItem]) -> bool {
+    if items.len() < 2 {
+        return false;
+    }
+
+    // Check for explicit tab separators
+    let tab_count = items.iter().filter(|i| i.text.contains('\t')).count();
+    if tab_count > 0 {
+        return true;
+    }
+
+    // Check for consistent spacing between items (column alignment indicator)
+    if items.len() >= 2 {
+        let mut gaps: Vec<i32> = Vec::new();
+        for i in 1..items.len() {
+            let gap = ((items[i].x - items[i - 1].x) * 100.0) as i32;
+            gaps.push(gap);
+        }
+
+        // If we have 3+ items and gaps are somewhat consistent, might be a table
+        if items.len() >= 3 {
+            let avg_gap: i32 = gaps.iter().sum::<i32>() / gaps.len() as i32;
+            let variance: i32 =
+                gaps.iter().map(|g| (g - avg_gap).abs()).sum::<i32>() / gaps.len() as i32;
+            // Low variance indicates column alignment
+            if variance < avg_gap / 4 && avg_gap > 50 {
+                return true;
+            }
+        }
+    }
+
+    // Check if items look like table cells (short, similar length)
+    let all_short = items
+        .iter()
+        .all(|i| i.text.len() <= 30 && !i.text.contains(' '));
+    let similar_length = items.len() >= 2 && {
+        let avg_len = items.iter().map(|i| i.text.len() as i32).sum::<i32>() / items.len() as i32;
+        items.iter().all(|i| {
+            let len = i.text.len() as i32;
+            (len - avg_len).abs() < avg_len / 2
+        })
+    };
+
+    all_short && similar_length && items.len() >= 3
+}
+
 /// Extract text items from PDF content stream with position information.
 ///
 /// Returns vector of (y_position, x_position, text) tuples.

core/src/composite_document_parser/tests.rs

@@ -2863,6 +2863,7 @@ fn extract_document_runtime_metadata_parses_ocr_block() {
                 ),
             ],
             metadata: None,
+            ..Default::default()
         };
 
     let metadata = extract_document_runtime_metadata(&doc).unwrap();

core/src/document/consume.rs

@@ -491,6 +491,7 @@ mod tests {
                 DocumentBlock::new(DocumentBlockKind::EmailHeader, Some("Subject"), "Hello"),
             ],
             metadata: None,
+            ..Default::default()
         };
 
         let lines = build_search_lines(&doc);
@@ -517,6 +518,7 @@ mod tests {
             .with_page(2)
             .with_ordinal(4)],
             metadata: None,
+            ..Default::default()
         };
 
         let metadata = llm_block_metadata(&doc, &[0], block_kind_label);
@@ -655,6 +657,8 @@ mod tests {
             max_chars: 8000,
             structural_summary: "\n## Structural Summary\n\nOverview\n",
             llm_answer: Some("Key finding"),
+            tables: &[],
+            pages: &[],
         };
 
         let built = build_parse_result(&input);
@@ -722,6 +726,8 @@ mod tests {
             max_chars: 8000,
             structural_summary: "\n## Structural Summary\n\nOverview\n",
             llm_answer: None,
+            tables: &[],
+            pages: &[],
         };
 
         let output = build_parse_tool_output(&input);

core/src/document/consume_search.rs

@@ -1058,6 +1058,7 @@ mod tests {
                 DocumentBlock::new(DocumentBlockKind::EmailHeader, Some("Subject"), "Hello"),
             ],
             metadata: None,
+            ..Default::default()
         };
 
         let chunks = vec![