TIMX 494 - yield deduped, most recent records

[ghukill]

Purpose and background context

This PR marks a somewhat important milestone in this library: support for reading only "current" records from the parquet dataset in a performant and memory-efficient manner.

This builds on PR #143, which introduced the TIMDEXRunManager which provided explicit lists of parquet files based on run ETL metadata. With that in hand, TIMDEXDataset was updated in two key ways.

First, the .load() method was updated to include a current_records:bool keyword argument, which when True, will utilize the TIMDEXRunManager to set an explicit list of parquet files for any future reading.

Second, when current_records are requested a private attribute is set TIMDEXDataset._dedupe_on_read = True. This informs all read methods to perform record deduping on read, which they can confidently and easily do with the dataset parquet files explicitly set in reverse chronological order.

The majority of the work was not getting this functional, but keeping the API minimal, intuitive, and non-surprising where possible. For example, to read only current dspace records from the dataset, only the following is required:

from timdex_dataset_api import TIMDEXDataset

timdex_dataset = TIMDEXDataset("s3://bucket/dataset")
timdex_dataset.load(current_records=True, source="dspace")

# from here, all read methods will produce only current records

The first anticipated use case is for TIM, where a new CLI command along the lines of reindex-source would utilize this new .load() + read technique to yield all and only current records for a source purely from dataset data. No re-harvests would be required. This is useful for development work, and could be a very useful option in disaster recovery situations.

How can a reviewer manually see the effects of these changes?

1- Set Dev Admin AWS credentials

2- Set following env vars (if not already):

TDA_LOG_LEVEL=DEBUG
WARNING_ONLY_LOGGERS=asyncio,botocore,urllib3,s3transfer,boto3

3- Start Ipython

pipenv run shell

4- Load the dataset for libguides current records only

from timdex_dataset_api import TIMDEXDataset
from timdex_dataset_api.config import configure_dev_logger

configure_dev_logger()

timdex_dataset = TIMDEXDataset("s3://ghukill-test/timdex-dataset/prod_2025_05_05")
timdex_dataset.load(current_records=True, source="libguides")

• Note some logging outputs that indicate the dataset is loaded and filtered

5- Given the small number of records for this source, load them all into a dataframe (takes ~5 seconds):

df = timdex_dataset.read_dataframe()

6- Inspecting the dataframe to confirm that timdex_record_id is unique:

df.timdex_record_id.is_unique
# Out[3]: True

7- Observe the ETL runs present in the records:

df[['run_date','run_type','run_id']].value_counts()

• The count is effectively how many records from that run are "current" (i.e. no new versions after that)
• Note the single run_type="full" has the bulk of the records, and run_type="daily" runs are just incrementally updating

8- Lastly, optional and just for kicks, an more full fledged example showing looping through all current records for dspace:

import time

from timdex_dataset_api import TIMDEXDataset
from timdex_dataset_api.config import configure_dev_logger

configure_dev_logger()

timdex_dataset = TIMDEXDataset("s3://ghukill-test/timdex-dataset/prod_2025_05_05")
timdex_dataset.load(current_records=True, source="dspace")

read_columns = ["timdex_record_id", "run_date", "action", "run_id"]
t0 = time.perf_counter()
count = 0
t1 = time.perf_counter()
for i, batch in enumerate(timdex_dataset.read_batches_iter(columns=read_columns)):
    df = batch.to_pandas()
    row = df.iloc[0]
    count += len(batch)
    print(
        f"batch {i}, rows {len(batch)}, total {count}, run_date {row.run_date}, "
        f"run_id {row.run_id}, elapsed {time.perf_counter() - t1}"
    )
    t1 = time.perf_counter()
print(f"total: {time.perf_counter()-t0}")

• Note the batches start with small numbers of records, just updates, until you finally get to the first run_type="full" where you get lots of records
• If you have resource monitoring open with something like btop, note that memory usage remains low

Includes new or updated dependencies?

NO

Changes expectations for external applications?

YES: applications like TIM may utilize new TDA functionality to read only current records

What are the relevant tickets?

• https://mitlibraries.atlassian.net/browse/TIMX-493
• https://mitlibraries.atlassian.net/browse/TIMX-494

Developer

• All new ENV is documented in README
• All new ENV has been added to staging and production environments
• All related Jira tickets are linked in commit message(s)
• Stakeholder approval has been confirmed (or is not needed)

Code Reviewer(s)

• The commit message is clear and follows our guidelines (not just this PR message)
• There are appropriate tests covering any new functionality
• The provided documentation is sufficient for understanding any new functionality introduced
• Any manual tests have been performed or provided examples verified
• New dependencies are appropriate or there were no changes

[coveralls]

Pull Request Test Coverage Report for Build 15189612409

Details

• 42 of 43 (97.67%) changed or added relevant lines in 3 files are covered.
• No unchanged relevant lines lost coverage.
• Overall coverage increased (+0.3%) to 94.872%

---

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
timdex_dataset_api/dataset.py	33	34	97.06%

Totals
Change from base Build 15168616734:	0.3%
Covered Lines:	296
Relevant Lines:	312

---

💛 - Coveralls

[ehanson8]

Looking good and works as expected, just a few questions!

[ghukill]

@jonavellecuerdo , @ehanson8 - I've stumbled on kind of an interesting wrinkle / edge case with this approach. @ehanson8, in a way, it's related to your comment about self.dataset = self._get_filtered_dataset(**filters) after we've already filtered for the current parquet files. It's relevant there, but also for any read method where do the same thing.

My preference would be to keep this PR as-is, which establishes some flags and methods that get us very close. In fact, it would work as needed for TIM to refresh a source as long as TIM was careful with how it uses this new functionality.

But, I think there is room for .load(current_records=True) to be less surprising, even if additional filtering is performed after the initial load. More on that in a follow-up PR, but wanted to note here in case it comes up during @jonavellecuerdo's review.

The very short example of this edge case:

• there are 100 "current" records
• the most recent "daily" run had 10 action=delete rows
• therefore, you have 90 records with action=index and 10 records with action=delete yielded as "current" records

While those action=delete rows are the current version of those records, we would not see them in Opensearch, and this challenges our definition of "current" for those records. Yes, it's the "current" version of the record from the dataset's POV, but it's not in Opensearch by virtue of that most recent version having action=delete.

I would propose that when load(current_records=True) is used you:

• always see only the most recent version of that record in the parquet dataset
• if filtering removes that particular version, the record is omitted entirely

[ehanson8]

I would propose that when load(current_records=True) is used you:

• always see only the most recent version of that record in the parquet dataset
• if filtering removes that particular version, the record is omitted entirely

I largely agree but did wonder if this could complicate troubleshooting if we can't find the ID of a particular record in the loaded records since it was deleted. However, I imagine load(current_records=True) will be primarily used for reindexing and not troubleshooting, so I think there is minimal risk if any risk at all

[ghukill]

I would propose that when load(current_records=True) is used you:

• always see only the most recent version of that record in the parquet dataset
• if filtering removes that particular version, the record is omitted entirely

I largely agree but did wonder if this could complicate troubleshooting if we can't find the ID of a particular record in the loaded records since it was deleted. However, I imagine load(current_records=True) will be primarily used for reindexing and not troubleshooting, so I think there is minimal risk if any risk at all

Yeah, it's interesting. I know that I probably overuse "ergonomics" but I think this is a really good example of it. Functionality wise it's all pretty simple, but it's about matching public interfaces, internal arguments and naming conventions, and behavior with a user's expectations.

My hope is that we could nudge any concept of "current records" towards 100% matching what we find in Opensearch. So if alma:1234 was most recently deleted from Opensearch as part of daily ETL run run-99, I see two concrete paths when you utilize current_records=True:

1. without any other filters, you encounter alma:1234, action=delete, run_id=run-99 which is the form of ETL record that caused the delete
2. you apply filtering that removes action=delete, or run_id=run-99, or maybe source=libguides, you should not see this record at all

I think the implementation details for this have a little complexity, but I think the more intuitive ergonomics of it will be worth it.

[ghukill]

@jonavellecuerdo, @ehanson8 - building on my comment above about an edge case, if interested, here is a unit test I'm working on locally that gets at it.

My goal is to work through a solution for this test, which currently fails, and have that be the basis of a followup PR.

def test_dataset_current_records_accurate_with_additional_filters(new_local_dataset):
    """Simulate ETL runs with the following flow:

        1. full run-1 establishes 5 records
            - current state = 5 to-index
        2. daily run-2 updates the first 5 and adds 5 new ones
            - current state = 10 to-index
        3. daily run-3 has 3 deletes
            a. current state = 7 to-index, 3 to-delete

    The interesting records here are alma:0|1|2 where the most recent version is
    action="delete".  If during a read method we filtered action="index", we would expect
    that we don't see alma:0|2 with action="delete".  However, we should *also* not see
    alma:0|1|2 with action="index" from run-2, as these are not "current" versions of the
    record.

    This test ensures that even if additional filtering is applied after
    load(current_records=True) is performed, we do not encounter non-current forms of a
    record.
    """
    run_params = [
        (5, "alma", "2025-01-01", "full", "index", "run-1"),
        (10, "alma", "2025-01-02", "daily", "index", "run-2"),
        (3, "alma", "2025-01-03", "daily", "delete", "run-3"),
    ]
    for params in run_params:
        num_records, source, run_date, run_type, action, run_id = params
        records = generate_sample_records(
            num_records,
            timdex_record_id_prefix=source,
            source=source,
            run_date=run_date,
            run_type=run_type,
            action=action,
            run_id=run_id,
        )
        new_local_dataset.write(records)

    # without any additional filtering, confirm current records are accurate
    new_local_dataset.load(current_records=True)
    assert new_local_dataset.read_dataframe().action.value_counts().to_dict() == {
        "index": 7,  # run-1 and run-2 established 10 records
        "delete": 3,  # run-3 had 3 deletes
    }

    # apply action="index" filtering on read, assert that alma:0|1|2 are not present
    new_local_dataset.load(current_records=True)
    current_to_index = new_local_dataset.read_dataframe(action="index")

    assert len(current_to_index) == 7
    for timdex_record_id in ["alma:0", "alma:1", "alma:2"]:
        assert timdex_record_id not in list(current_to_index.timdex_record_id)

[ehanson8]

Makes sense to me!

[jonavellecuerdo]

One small request that will likely be addressed in the follow-up PR!

[jonavellecuerdo]

Can you update the docstring for TIMDEXDataset.load to include a description for the new current_records: bool keyword arg?

[ghukill]

It is coming in the followup PR! Good catch.