current_records view relies on temp table

Why these changes are being introduced:

The current_records view has always been a performance and resource
bottleneck.  Moving to metadata and SQL has helped, but there was a little
kink left in relation to using that metadata for data retreival.

We often would materialize a query to a pandas dataframe for to use
to drive data retrieval.  In that moment, we do not benefit from having
current_records be a view, when we're going to materialize the data
anyhow.

How this addresses that need:

Utilizing a DuckDB temp table, we take a small performance hit on
TIMDEXDatasetMetadata load, but then have near instant current_records
queries thereafter.

Additionally, we remove ordering in the metadata query for data
retrieval and perform this in-memory with the pandas dataframe.  Often
this may be quite small, but even if large, it's more efficient here
and already in python memory.

This will also set the stage for performing just-in-time metadata
queries as chunks before data retrieval, versus pulling all metadata
rows in one query and then chunking that in memory.

Side effects of this change:
* Quicker metadata queries, small performance hit on load.  Appears
similarly memory intensive.

Relevant ticket(s):
* https://mitlibraries.atlassian.net/browse/TIMX-543

Author: ghukill

tests/test_read.py

@@ -125,8 +125,8 @@ def test_read_batches_where_and_dataset_filters_are_combined(timdex_dataset_mult
     [
         "SELECT * FROM current_records WHERE source = 'libguides'",
         "FROM records WHERE source = 'libguides'",
-        "source = 'libguides';",
-        " run_date = '2024-12-01';  ",
+        "ORDER BY timdex_record_id",
+        "LIMIT 3",
     ],
 )
 def test_read_batches_where_rejects_non_predicate_sql(

timdex_dataset_api/dataset.py

@@ -375,10 +375,13 @@ def read_batches_iter(
             key/value DatasetFilters
             - filters: simple filtering based on key/value pairs from DatasetFilters
         """
+        start_time = time.perf_counter()
+
         # build and execute metadata query
         metadata_time = time.perf_counter()
         meta_query = self.metadata.build_meta_query(table, where, **filters)
         meta_df = self.metadata.conn.query(meta_query).to_df()
+        meta_df = meta_df.sort_values(by=["filename", "run_record_offset"])
         logger.debug(
             f"Metadata query identified {len(meta_df)} rows, "
             f"across {len(meta_df.filename.unique())} parquet files, "
@@ -410,6 +413,10 @@ def read_batches_iter(
                 f"@ {batch_rps} records/second, total yielded: {total_yield_count}"
             )
 
+        logger.debug(
+            f"read_batches_iter() elapsed: {round(time.perf_counter()-start_time, 2)}s"
+        )
+
     def _iter_meta_chunks(self, meta_df: pd.DataFrame) -> Iterator[pd.DataFrame]:
         """Utility method to yield chunks of metadata query results."""
         for start in range(0, len(meta_df), self.config.duckdb_join_batch_size):

timdex_dataset_api/metadata.py

@@ -334,6 +334,7 @@ def setup_duckdb_context(self) -> DuckDBPyConnection:
         start_time = time.perf_counter()
 
         conn = duckdb.connect()
+        conn.execute("""SET enable_progress_bar = false;""")
         self.configure_duckdb_connection(conn)
 
         if not self.database_exists():
@@ -436,36 +437,67 @@ def _create_current_records_view(self, conn: DuckDBPyConnection) -> None:
 
         This view builds on the table `records`.
 
-        This view includes only the most current version of each record in the dataset.
-        Because it includes the `timdex_record_id` and `run_id`, it makes yielding the
-        current version of a record via a TIMDEXDataset instance trivial: for any given
-        `timdex_record_id` if the `run_id` doesn't match, it's not the current version.
+        This metadata view includes only the most current version of each record in the
+        dataset.  With the metadata provided from this view, we can streamline data
+        retrievals in TIMDEXDataset read methods.
         """
         logger.info("creating view of current records metadata")
 
-        query = f"""
-        create or replace view metadata.current_records as
-        with ranked_records as (
+        conn.execute(
+            """
+            set temp_directory = '/tmp';
+            """
+        )
+
+        conn.execute(
+            """
+            -- create temp table with current records using CTEs
+            create or replace temp table temp.main.current_records as
+            with
+                -- CTE of run_timestamp for last source full run
+                cr_source_last_full as (
+                    select
+                        source,
+                        max(run_timestamp) as last_full_ts
+                    from metadata.records
+                    where run_type = 'full'
+                    group by source
+                ),
+
+                -- CTE of all records, per source, on or after last full run
+                cr_since_last_full as (
+                    select
+                        r.*
+                    from metadata.records r
+                    join cr_source_last_full f using (source)
+                    where r.run_timestamp >= f.last_full_ts
+                ),
+
+                -- CTE of records ranked by run_timestamp, with tie breaker
+                cr_ranked_records as (
+                    select
+                        r.*,
+                        row_number() over (
+                            partition by r.source, r.timdex_record_id
+                            order by
+                                r.run_timestamp desc nulls last,
+                                r.run_id desc nulls last,
+                                r.run_record_offset desc nulls last
+                        ) as rn
+                    from cr_since_last_full r
+                )
+
+            -- final select for current records (rn = 1)
             select
-                r.*,
-                row_number() over (
-                    partition by r.timdex_record_id
-                    order by r.run_timestamp desc
-                ) as rn
-            from metadata.records r
-            where r.run_timestamp >= (
-                select max(r2.run_timestamp)
-                from metadata.records r2
-                where r2.source = r.source
-                and r2.run_type = 'full'
-            )
+                * exclude (rn)
+            from cr_ranked_records
+            where rn = 1;
+
+            -- create view in metadata schema
+            create or replace view metadata.current_records as
+            select * from temp.main.current_records;
+            """
         )
-        select
-            {','.join(ORDERED_METADATA_COLUMN_NAMES)}
-        from ranked_records
-        where rn = 1;
-        """
-        conn.execute(query)
 
     def merge_append_deltas(self) -> None:
         """Merge append deltas into the static metadata database file."""
@@ -602,7 +634,6 @@ def build_meta_query(
         ).select_from(sa_table)
         if combined is not None:
             stmt = stmt.where(combined)
-        stmt = stmt.order_by(sa_table.c.filename, sa_table.c.run_record_offset)
 
         # using DuckDB dialect, compile to SQL string
         compiled = stmt.compile(