feat(ingestion/s3): ignore depth mismatched path

[eagle-25]

Summary

Feature Addition:

Ignores the paths whose depth dose not match to path_spec.include while ingesting S3.

Changes

As-is

• Retrieves all paths under the prefix.

To-be

• Retrieves only paths whose depth matches path_spec.include.

Background

• Warning messages were triggered during S3 ingestion.
• This issue was caused by attempting to ingest paths whose depth does not match path_spec.include.
• Since this behavior seems unintended, I propose adding this feature.

Logs

path_spec: s3://bucket/{table}/{partition0}/*.csv

s3_files:
1. s3://bucket/shop_log/year=2025/month=01/day=01/0000.csv
2. s3://bucket/shop_log/year=2025/month=01/day=02/0000.csv
3. s3://bucket/user_log/2025/0000.csv

============

**Before**

$ python s3_ingestion.py

INFO - Processing folder: datahub-test/shop_log
WARNING - Unable to find any files in the folder datahub-test/shop_log/year=2025. Skipping...
WARNING - Unable to find any files in the folder datahub-test/shop_log/year=2025/month=01. Skipping...
WARNING - Unable to find any files in the folder datahub-test/shop_log/year=2025/month=01/day=01. Skipping...
WARNING - Unable to find any files in the folder datahub-test/shop_log/year=2025/month=01/day=02. Skipping...
WARNING - Unable to find any files in the folder s3://bucket/datahub-test/shop_log/year=2025/. Skipping...


INFO - Processing folder: datahub-test/user_log
INFO - Getting files from folder: s3://bucket/datahub-test/user_log/2025/
INFO - Extracting table schema from file: s3://bucket/datahub-test/user_log/2025/ai_kpi.04_kst_v4_dag_runs.csv
INFO - Creating dataset urn with name: bucket/datahub-test/user_log

============

**After**

$ python s3_ingestion.py

INFO - Processing folder: datahub-test/shop_log
INFO - Getting files from folder: s3://bucket/datahub-test/shop_log/year=2025/
WARNING - Unable to find any files in the folder s3://bucket/datahub-test/shop_log/year=2025/. Skipping...


INFO - Processing folder: datahub-test/user_log
INFO - Getting files from folder: s3://bucket/datahub-test/user_log/2025/
INFO - Extracting table schema from file: s3://bucket/datahub-test/user_log/2025/ai_kpi.04_kst_v4_dag_runs.csv
INFO - Creating dataset urn with name: bucket/datahub-test/user_log

How to reproduce

You can reproduce this issue by running the code below.

# s3_ingestion.py

from datahub.ingestion.run.pipeline import Pipeline

_DATAHUB_GMS_ENDPOINT = "<your_gms_endpoint>"
_DATAHUB_ACCESS_TOKEN = "<your_token>"

_S3_PATH = "s3://<your-bucket>/datahub-test/{table}/{partition0}/*.csv"

s3_config = {
        "source": {
            "type": "s3",
            "config": {
                "path_spec": {
                    "include": _S3_PATH
                },
                "aws_config": {
                    "aws_region": "ap-northeast-2",
                },
            },
        },
        "sink": {
            "type": "datahub-rest",
            "config": {
                "server": _DATAHUB_GMS_ENDPOINT,
                "token": _DATAHUB_ACCESS_TOKEN,
            },
        },
    }

pipeline = Pipeline.create(s3_config)
pipeline.run()

Checklist

• The PR conforms to DataHub's Contributing Guideline (particularly Commit Message Format)
• Links to related issues (if applicable)
• Tests for the changes have been added/updated (if applicable)
• Docs related to the changes have been added/updated (if applicable). If a new feature has been added a Usage Guide has been added for the same.
• For any breaking change/potential downtime/deprecation/big changes an entry has been made in Updating DataHub

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Files with missing lines	Coverage Δ
...ngestion/src/datahub/ingestion/source/s3/source.py	86.53% <100.00%> (-0.35%)	⬇️

... and 5 files with indirect coverage changes

---

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update f4c0927...3d5d49b. Read the comment docs.

[eagle-25]

I added this function to log when a file is ignored.

If you have a better approach, please feel free to suggest it. 🙏

[treff7es]

You can add this to the report as well -> something like this ->

datahub/metadata-ingestion/src/datahub/ingestion/source/redshift/redshift.py

Line 651 in 96758e2

self.report.report_dropped(datahub_dataset_name)

[treff7es]

Just make sure in the report you collect in a lossy list

[eagle-25]

Added with a test case.

[eagle-25]

To clarify my intention, I wrote it as qna format.

Q: Why path_spec.allowed moved to L879?
A: Since path_spec.allowed already implements depth comparison, I used it to determine whether each path should be skipped or not while iterating over s3_objects.

Q: Does moving this logic to L879 make any breaking changes?
A: No, because it retains the original behavior by continuing to skip each disallowed path, even after being moved to L879.

[eagle-25]

Changes

• Add dropped files to the report.filtered
• Add description to each assertion