fix: support big csv files on langflow-less ingestion

[lucaseduoli]

This pull request enhances the document chunking and embedding workflow to ensure that text chunks sent for embedding are properly aligned with their embeddings, particularly by splitting oversized chunks before batching. It also improves the robustness of table extraction and adds comprehensive unit tests for the new chunk-splitting logic and edge cases in document processing.

Document chunking and embedding alignment:

• Introduced a new function split_chunks_by_max_tokens in src/utils/document_processing.py to pre-split any chunk whose token count exceeds a model-specific limit, ensuring that each chunk aligns 1-to-1 with its embedding. This function uses the appropriate tokenizer based on the embedding model.
• Integrated split_chunks_by_max_tokens into the document processing pipeline in src/models/processors.py, so that chunks are split before batching for embeddings, preventing misalignment between chunks and their embeddings. [1] [2]

Robustness improvements in table extraction:

• Updated extract_relevant in src/utils/document_processing.py to safely handle cases where table data is missing (None), preventing errors during extraction.

Testing:

• Added unit tests in tests/unit/test_document_processing_resplit.py for split_chunks_by_max_tokens, covering both under-limit and over-limit scenarios, and for handling tables with missing data in extract_relevant.

Refactoring and cleanup:

• Removed an unused variable (texts) in process_document_standard after refactoring the chunk splitting logic.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: b94be049-8c57-4d03-8083-f2bdde8cff5a

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix/csv_langflowless_ingestion

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[edwinjosechittilappilly]

LGTM

[lucaseduoli]

@coderabbitai please create a PR against this fixing the errors that caused the integration test failures

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

repo="langflow-ai/openrag"
pr="1966"

echo "== PR checks =="
gh pr checks "$pr" --repo "$repo" || true

echo
echo "== PR view (head/base, changed files summary) =="
gh pr view "$pr" --repo "$repo" --json number,title,headRefName,baseRefName,author,changedFiles,files,statusCheckRollup \
  --jq '{
    number,
    title,
    headRefName,
    baseRefName,
    author: .author.login,
    changedFiles,
    files: [.files[].path],
    checks: [.statusCheckRollup[]? | {
      name: (.name // .context // "unknown"),
      status: .status,
      conclusion: .conclusion,
      detailsUrl: .detailsUrl
    }]
  }'

echo
echo "== Locate candidate files =="
fd -a 'document_processing.py|processors.py|test_document_processing_resplit.py|conftest.py|integration' . || true

echo
echo "== AST outline: src/utils/document_processing.py =="
ast-grep outline src/utils/document_processing.py --view expanded || true

echo
echo "== AST outline: src/models/processors.py =="
ast-grep outline src/models/processors.py --view expanded || true

echo
echo "== Relevant slices: src/utils/document_processing.py =="
sed -n '1,260p' src/utils/document_processing.py || true

echo
echo "== Relevant slices: src/models/processors.py =="
sed -n '1,420p' src/models/processors.py || true

echo
echo "== New/related tests =="
sed -n '1,260p' tests/unit/test_document_processing_resplit.py || true

echo
echo "== Search for split_chunks_by_max_tokens usages =="
rg -n -C2 '\bsplit_chunks_by_max_tokens\s*\(' src tests || true

echo
echo "== Search for integration tests mentioning document processing / ingestion =="
rg -n -C2 --iglob '*integration*' 'document_processing|process_document_standard|extract_relevant|split_chunks_by_max_tokens|embedding' tests || true

Length of output: 38801

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

repo="langflow-ai/openrag"
run_id="28472586766"
job_id="84388865447"

echo "== Failed job metadata =="
gh api "repos/$repo/actions/jobs/$job_id" | jq '{name, status, conclusion, started_at, completed_at, html_url, steps: [.steps[] | {name, status, conclusion, number}]}'

echo
echo "== Failed job annotations =="
gh api "repos/$repo/check-runs/$job_id/annotations" || true

echo
echo "== Attempt to fetch run/job logs =="
tmpdir="$(mktemp -d)"
if gh run view "$run_id" --repo "$repo" --log > "$tmpdir/run.log" 2>/dev/null; then
  echo "-- First 250 lines of run log --"
  sed -n '1,250p' "$tmpdir/run.log"
else
  echo "gh run view --log unavailable; trying gh api for logs URL"
  logs_url=$(gh api "repos/$repo/actions/jobs/$job_id" --jq '.logs_url' || true)
  echo "logs_url=${logs_url:-<none>}"
fi

echo
echo "== Integration test file outline =="
ast-grep outline tests/integration/core/test_non_langflow_ingestion.py --view expanded || true

echo
echo "== Integration test file contents =="
sed -n '1,260p' tests/integration/core/test_non_langflow_ingestion.py || true

echo
echo "== PR diff for changed files =="
gh pr diff 1966 --repo "$repo" -- src/models/processors.py src/utils/document_processing.py tests/integration/core/test_non_langflow_ingestion.py tests/unit/test_document_processing_resplit.py | sed -n '1,320p'

Length of output: 37715

---

🏁 Script executed:

#!/bin/bash
set -euo pipefail

repo="langflow-ai/openrag"
run_id="28472586766"

tmpdir="$(mktemp -d)"
logfile="$tmpdir/run.log"

echo "== Download full run log =="
gh run view "$run_id" --repo "$repo" --log > "$logfile"

echo
echo "== Pytest failure summary from log =="
rg -n -C4 'FAILED|FAILURES|ERROR|AssertionError|E +assert|traceback|short test summary info|test_non_langflow_csv_ingestion_with_splitting' "$logfile" || true

echo
echo "== Tail of run log =="
tail -n 220 "$logfile" || true

echo
echo "== Search for upload_ingest and document file processor path =="
rg -n -C3 '/router/upload_ingest|class DocumentFileProcessor|process_item\(|process_document_standard\(' src tests || true

echo
echo "== Focused slices around DocumentFileProcessor and upload flow =="
sed -n '460,760p' src/models/processors.py || true

echo
echo "== Git refs for current PR branch vs base =="
git rev-parse HEAD
git branch --show-current || true
git remote -v | head

echo
echo "== Focused git diff for relevant files =="
git diff -- src/models/processors.py src/utils/document_processing.py tests/integration/core/test_non_langflow_ingestion.py tests/unit/test_document_processing_resplit.py | sed -n '1,360p' || true

Length of output: 147660

---

@lucaseduoli I can’t open/push a PR directly from here, but I did trace the failing integration job and the failures are not coming from the new CSV chunk-splitting path.

What’s failing in #1966:

• tests/integration/core/test_runtime_migration.py::test_legacy_file_state_migrates_on_backend_startup_and_is_idempotent
  • failing with 403 on /chat/history
  • required permission: conversations:read:own
• tests/integration/core/test_startup_ingest.py::test_startup_ingest_creates_task[True]
• tests/integration/core/test_startup_ingest.py::test_startup_ingest_creates_task[False]
  • failing with: DB engine not yet initialized. RBACService called before lifespan startup.

Notably:

• tests/integration/core/test_non_langflow_ingestion.py::test_non_langflow_csv_ingestion_with_splitting passed

So the actionable fix is around lifespan/startup initialization and RBAC expectations/permissions in the migration test, not the new document chunking logic.

• Fix the startup-ingest integration tests so the FastAPI app lifespan fully runs before readiness checks hit /search
• Fix the runtime-migration integration test so it uses a user/session state that has conversations:read:own, or update the migrated permission seeding path if that permission is missing after startup
• Keep the CSV/non-Langflow ingestion changes intact, since that integration test is already passing
• Re-run the three failing integration tests first, then the full tests workflow

---

Failed to handle agent chat message. Please try again.