diff --git a/sds_data_manager/lambda_code/SDSCode/api_lambdas/release_api.py b/sds_data_manager/lambda_code/SDSCode/api_lambdas/release_api.py
index ecd137c65..3c182d28b 100644
--- a/sds_data_manager/lambda_code/SDSCode/api_lambdas/release_api.py
+++ b/sds_data_manager/lambda_code/SDSCode/api_lambdas/release_api.py
@@ -3,19 +3,17 @@
 import datetime
 import json
 import logging
-from pathlib import Path
 
 import imap_data_access
 from imap_data_access.file_validation import (
-    AncillaryFilePath,
-    ScienceFilePath,
     generate_imap_file_path,
 )
-from sqlalchemy import func, or_, text, union_all
+from sqlalchemy import and_, func, literal, or_, select
 
 from ..database import database as db
 from ..database import models
 from ..spice_utilities import download_from_s3
+from .utils import build_latest_version_query
 
 # Logger setup
 logger = logging.getLogger(__name__)
@@ -63,27 +61,9 @@ def validate_query_params(event):
             ),
         }
 
-    if release_type == "release" and "release_number" not in query_params:
-        return {
-            "statusCode": 400,
-            "body": json.dumps(
-                "'release_number' query parameter is required when "
-                "'release_type' is 'release'. Please provide a release_number "
-                "indicating which release batch to apply. For example, "
-                "withhold files with 'release_number=1' will be included in "
-                "the first release batch, 'release_number=2' in the second, "
-                "and so on."
-            ),
-        }
-
     valid_parameters = [
-        "instrument",
-        "start_date",
-        "end_date",
         "release_type",
-        "exclude_file",
         "manifest_file",
-        "release_number",
     ]
 
     for param in query_params:
@@ -105,182 +85,47 @@ def validate_query_params(event):
     }
 
 
-def query_latest_science_files(
-    session, instrument, start_date, end_date, science_files_to_exclude=None
-):
-    """Query for the latest-version science file paths matching given criteria.
-
-    Latest version is determined by finding the maximum major_version, then
-    within that, the maximum minor_version for each unique file grouping.
-    """
-    science_table = models.ScienceFiles
-
-    # Check if any file in the range has a non-null repointing
-    has_repointing = (
-        session.query(science_table)
-        .filter(
-            science_table.instrument == instrument,
-            science_table.start_date >= start_date,
-            science_table.start_date <= end_date,
-            science_table.repointing.isnot(None),
-        )
-        .first()
-        is not None
-    )
+def parse_manifest_line(line: str):
+    """Parse a release manifest line into parts.
 
-    if has_repointing:
-        # Find max major_version for each file group
-        max_major_subq = (
-            session.query(
-                science_table.instrument,
-                science_table.data_level,
-                science_table.descriptor,
-                science_table.start_date,
-                science_table.repointing,
-                func.max(science_table.major_version).label("max_major"),
-            )
-            .group_by(
-                science_table.instrument,
-                science_table.data_level,
-                science_table.descriptor,
-                science_table.start_date,
-                science_table.repointing,
-            )
-            .subquery()
-        )
-
-        # Within max major_version, find max minor_version
-        max_minor_subq = (
-            session.query(
-                science_table.instrument,
-                science_table.data_level,
-                science_table.descriptor,
-                science_table.start_date,
-                science_table.repointing,
-                science_table.major_version,
-                func.max(science_table.minor_version).label("max_minor"),
-            )
-            .join(
-                max_major_subq,
-                (science_table.instrument == max_major_subq.c.instrument)
-                & (science_table.data_level == max_major_subq.c.data_level)
-                & (science_table.descriptor == max_major_subq.c.descriptor)
-                & (science_table.start_date == max_major_subq.c.start_date)
-                & (science_table.repointing == max_major_subq.c.repointing)
-                & (science_table.major_version == max_major_subq.c.max_major),
-            )
-            .group_by(
-                science_table.instrument,
-                science_table.data_level,
-                science_table.descriptor,
-                science_table.start_date,
-                science_table.repointing,
-                science_table.major_version,
-            )
-            .subquery()
-        )
-
-        latest_science_files = (
-            session.query(science_table)
-            .join(
-                max_minor_subq,
-                (science_table.instrument == max_minor_subq.c.instrument)
-                & (science_table.data_level == max_minor_subq.c.data_level)
-                & (science_table.descriptor == max_minor_subq.c.descriptor)
-                & (science_table.start_date == max_minor_subq.c.start_date)
-                & (science_table.repointing == max_minor_subq.c.repointing)
-                & (science_table.major_version == max_minor_subq.c.major_version)
-                & (science_table.minor_version == max_minor_subq.c.max_minor),
-            )
-            .filter(
-                science_table.instrument == instrument,
-                science_table.start_date >= start_date,
-                science_table.start_date <= end_date,
-            )
-        )
-    else:
-        # Find max major_version for each file group
-        max_major_subq = (
-            session.query(
-                science_table.instrument,
-                science_table.data_level,
-                science_table.descriptor,
-                science_table.start_date,
-                func.max(science_table.major_version).label("max_major"),
-            )
-            .group_by(
-                science_table.instrument,
-                science_table.data_level,
-                science_table.descriptor,
-                science_table.start_date,
-            )
-            .subquery()
-        )
+    Parameters
+    ----------
+    line : str
+        A single non-empty manifest line.
 
-        # Within max major_version, find max minor_version
-        max_minor_subq = (
-            session.query(
-                science_table.instrument,
-                science_table.data_level,
-                science_table.descriptor,
-                science_table.start_date,
-                science_table.major_version,
-                func.max(science_table.minor_version).label("max_minor"),
-            )
-            .join(
-                max_major_subq,
-                (science_table.instrument == max_major_subq.c.instrument)
-                & (science_table.data_level == max_major_subq.c.data_level)
-                & (science_table.descriptor == max_major_subq.c.descriptor)
-                & (science_table.start_date == max_major_subq.c.start_date)
-                & (science_table.major_version == max_major_subq.c.max_major),
-            )
-            .group_by(
-                science_table.instrument,
-                science_table.data_level,
-                science_table.descriptor,
-                science_table.start_date,
-                science_table.major_version,
-            )
-            .subquery()
+    Returns
+    -------
+    tuple[str, str, str, bool] | None
+        Parsed ``(instrument, data_type, descriptor, release_flag)`` or
+        ``None`` if the line does not contain exactly four comma-separated
+        fields.
+    """
+    # Skip comment or empty lines
+    if line.startswith("#") or line.strip() == "":
+        return None
+    # Skip Header
+    if line.startswith("instrument,"):
+        return None
+
+    parts = [item.strip() for item in line.split(",")]
+    if len(parts) != 4:
+        raise ValueError(
+            f"Manifest line must contain exactly four comma-separated fields: {line}"
         )
 
-        latest_science_files = (
-            session.query(science_table)
-            .join(
-                max_minor_subq,
-                (science_table.instrument == max_minor_subq.c.instrument)
-                & (science_table.data_level == max_minor_subq.c.data_level)
-                & (science_table.descriptor == max_minor_subq.c.descriptor)
-                & (science_table.start_date == max_minor_subq.c.start_date)
-                & (science_table.major_version == max_minor_subq.c.major_version)
-                & (science_table.minor_version == max_minor_subq.c.max_minor),
-            )
-            .filter(
-                science_table.instrument == instrument,
-                science_table.start_date >= start_date,
-                science_table.start_date <= end_date,
-            )
-        )
-    if science_files_to_exclude:
-        latest_science_files = latest_science_files.filter(
-            ~science_table.file_path.in_(science_files_to_exclude)
-        )
-    results = list(latest_science_files)
-    logger.info(f"Found {len(results)} science file(s) for instrument={instrument}")
-    return results
+    instrument, data_type, descriptor, release_flag = parts
+    return instrument, data_type, descriptor, release_flag.lower() == "true"
 
 
-def get_latest_ancillary_files(
+def latest_ancillary_release(
     session,
-    instrument: str,
     start_date: datetime.datetime,
     end_date: datetime.datetime,
-    ancillary_files_to_exclude: list | None = None,
-) -> list:
-    """Get latest-version ancillary files for an instrument over a date range.
+    line: str,
+):
+    """Set the released flag to True for latest-version ancillary files.
 
-    The function retrieves files in two groups based on overlap with date range:
+    The function selects files in two groups based on overlap with date range:
 
         Files with explicit end_date in their filename:
             overlaps if start_date <= query_end AND end_date >= query_start
@@ -294,129 +139,187 @@ def get_latest_ancillary_files(
     ----------
     session : orm session
         Database session.
-    instrument : str
-        Instrument name.
     start_date : datetime.datetime
         Start of query date range.
     end_date : datetime.datetime
         End of query date range.
-    ancillary_files_to_exclude : list, optional
-        List of ancillary file paths to exclude from results, by default None
+    line : str
+        Manifest line describing the ancillary release selection.
 
     Returns
     -------
     list
-        List of file paths ordered by file_path.
+        A list of the latest version ancillary files released.
     """
     ancillary_table = models.AncillaryFiles
+    instrument, data_type, descriptor, _ = parse_manifest_line(line)
+    # Scenarios:
+    # hit, all, all, true, -- release all ancillary files
+    # hit, ancillary, all, true -- release all ancillary descriptors
+    # hit, ancillary, x-descriptor, true - release only specified
+    if data_type == "all" or descriptor == "all":
+        filters = [ancillary_table.instrument == instrument]
+    else:
+        filters = [
+            ancillary_table.instrument == instrument,
+            ancillary_table.descriptor == descriptor,
+        ]
+
+    #
+    # Filter first so window functions operate on a small dataset.
+    #
+    filtered = select(ancillary_table).where(*filters).subquery()
 
-    # Step 1: Get latest version per (descriptor, start_date, end_date)
-    # Filter by instrument early to reduce data processed
-    row_num_col = (
-        func.row_number()
+    #
+    # RANK() to get latest version.
+    #
+    version_rank = (
+        func.rank()
         .over(
             partition_by=[
-                ancillary_table.descriptor,
-                ancillary_table.start_date,
-                ancillary_table.end_date,
+                filtered.c.instrument,
+                filtered.c.descriptor,
+                filtered.c.start_date,
+                filtered.c.end_date,
             ],
-            order_by=ancillary_table.version.desc(),
+            order_by=filtered.c.version.desc(),
         )
-        .label("row_num")
+        .label("version_rank")
     )
 
-    latest_versions = (
-        session.query(
-            ancillary_table.file_path,
-            ancillary_table.descriptor,
-            ancillary_table.start_date,
-            ancillary_table.end_date,
+    # LEAD() to set start_date as end_date for files that didn't
+    # end_date
+    next_start_date = (
+        func.lead(filtered.c.start_date)
+        .over(
+            partition_by=[
+                filtered.c.instrument,
+                filtered.c.descriptor,
+            ],
+            order_by=filtered.c.start_date,
         )
-        .filter(ancillary_table.instrument == instrument)
-        .add_columns(row_num_col)
-        .subquery()
+        .label("next_start_date")
     )
 
-    latest = (
-        session.query(
-            latest_versions.c.file_path,
-            latest_versions.c.descriptor,
-            latest_versions.c.start_date,
-            latest_versions.c.end_date,
-        )
-        .filter(latest_versions.c.row_num == 1)
-        .subquery()
+    windowed = select(
+        filtered,
+        version_rank,
+        next_start_date,
+    ).subquery()
+
+    #
+    # Keep only the latest major version.
+    #
+    latest = select(windowed).where(windowed.c.version_rank == 1).subquery()
+
+    #
+    # Use end_date when present; otherwise use the next file's
+    # start_date. If there is no next file, treat it as open-ended.
+    #
+    next_end = func.coalesce(latest.c.next_start_date, literal(datetime.datetime.max))
+
+    overlap_condition = or_(
+        and_(latest.c.end_date.is_not(None), latest.c.end_date >= start_date),
+        and_(latest.c.end_date.is_(None), next_end > start_date),
     )
 
-    # Step 2: Files WITH end_date - simple overlap check
-    with_end_date_query = session.query(latest.c.file_path).filter(
-        latest.c.end_date.isnot(None),
+    latest_file_paths = select(latest.c.file_path).where(
         latest.c.start_date <= end_date,
-        latest.c.end_date >= start_date,
+        overlap_condition,
     )
-
-    # Step 3: Files WITHOUT end_date - use LEAD() to find coverage end
-    next_start_col = (
-        func.lead(latest.c.start_date)
-        .over(partition_by=latest.c.descriptor, order_by=latest.c.start_date)
-        .label("next_start_date")
+    latest_records = session.query(models.AncillaryFiles).filter(
+        models.AncillaryFiles.file_path.in_(latest_file_paths)
     )
+    release_rows = latest_records.all()
 
-    no_end_with_next = (
-        session.query(latest.c.file_path, latest.c.start_date, next_start_col)
-        .filter(latest.c.end_date.is_(None))
-        .subquery()
+    latest_records.update(
+        {models.AncillaryFiles.released: True},
+        synchronize_session=False,
     )
+    logger.info(f"Released {len(release_rows)} ancillary files")
+    return release_rows
 
-    no_end_date_query = session.query(no_end_with_next.c.file_path).filter(
-        no_end_with_next.c.start_date <= end_date,
-        or_(
-            no_end_with_next.c.next_start_date.is_(None),
-            no_end_with_next.c.next_start_date > start_date,
-        ),
-    )
 
-    # Combine — ORDER BY positional index works across all DB backends for UNION ALL
-    combined = union_all(with_end_date_query, no_end_date_query).order_by(text("1"))
+def latest_science_release(session, start_date, end_date, line):
+    """Set the released flag to True for latest-version science files.
 
-    latest_ancillary_files = [row[0] for row in session.execute(combined).fetchall()]
+    Parameters
+    ----------
+    session : orm session
+        Database session.
+    start_date : datetime.datetime
+        Start of query date range.
+    end_date : datetime.datetime
+        End of query date range.
+    line : str
+        Manifest line describing the science release selection.
 
-    # Now exclude any files in the exclude list
-    if ancillary_files_to_exclude:
-        latest_ancillary_files = [
-            p for p in latest_ancillary_files if p not in ancillary_files_to_exclude
+    Returns
+    -------
+    list
+        A list of the latest version science files released.
+    """
+    sci = models.ScienceFiles.__table__.c
+    instrument, data_type, descriptor, _ = parse_manifest_line(line)
+
+    # Construct query logic based on different scenarios:
+    # 1. hit, all, all, true -- release all data levels
+    if data_type == "all":
+        query = [
+            sci.instrument == instrument,
+            sci.start_date >= start_date,
+            sci.start_date <= end_date,
+        ]
+    # 2. hit, l1a, all, true -- release all descriptor for given level
+    elif descriptor == "all":
+        query = [
+            sci.instrument == instrument,
+            sci.data_level == data_type,
+            sci.start_date >= start_date,
+            sci.start_date <= end_date,
+        ]
+    # 3. release specified level and descriptor
+    else:
+        query = [
+            sci.instrument == instrument,
+            sci.data_level == data_type,
+            sci.descriptor == descriptor,
+            sci.start_date >= start_date,
+            sci.start_date <= end_date,
         ]
 
-    if not latest_ancillary_files:
-        logger.info(f"Found 0 ancillary file(s) for instrument={instrument}")
-        return []
-
-    results = list(
-        session.query(models.AncillaryFiles).filter(
-            models.AncillaryFiles.file_path.in_(latest_ancillary_files)
-        )
+    latest = build_latest_version_query(
+        filters=query,
     )
-    logger.info(f"Found {len(results)} ancillary file(s) for instrument={instrument}")
-    return results
+    latest_file_paths = latest.with_only_columns(latest.selected_columns.file_path)
+    latest_records = session.query(models.ScienceFiles).filter(
+        models.ScienceFiles.file_path.in_(latest_file_paths)
+    )
+    release_rows = latest_records.all()
+    # Finally update released flag to True
+    latest_records.update(
+        {models.ScienceFiles.released: True},
+        synchronize_session=False,
+    )
+    logger.info(f"Released {len(release_rows)} science files")
+    return release_rows
 
 
-def download_read_file(exception_list_file_path):
-    """Download a manifest file from S3 and group its entries by file type.
+def download_file(manifest_file):
+    """Download a manifest file from S3.
 
     Parameters
     ----------
-    exception_list_file_path : str
+    manifest_file : str
         S3 path to the manifest text file. Each line is an IMAP file path.
 
     Returns
     -------
-    tuple[list[str], list[str]]
-        A tuple of (science_files, ancillary_files) where each entry is the
-        file path string listed in the manifest.
+    Path
+        Download file path
     """
     # Create the proper file path object based on the extension and filename
-    file_path = Path(exception_list_file_path)
-    path_obj = generate_imap_file_path(file_path.name)
+    path_obj = generate_imap_file_path(manifest_file)
 
     s3_file_path = (
         path_obj.construct_path()
@@ -427,178 +330,74 @@ def download_read_file(exception_list_file_path):
     logger.debug(f"Downloading manifest file from S3 path: {s3_file_path}")
     download_path = download_from_s3(s3_file_path)
     logger.debug(f"Local path after download: {download_path}")
-    lines = download_path.read_text(encoding="utf-8").splitlines()
-
-    science_files = []
-    ancillary_files = []
-    for line in lines:
-        filename = line.strip()
-        if not filename:
-            continue
-        file_obj = imap_data_access.file_validation.generate_imap_file_path(filename)
-        if isinstance(file_obj, ScienceFilePath):
-            science_files.append(filename)
-        elif isinstance(file_obj, AncillaryFilePath):
-            ancillary_files.append(filename)
-        else:
-            logger.warning(f"Unrecognized file type in manifest, skipping: {filename}")
-
-    return science_files, ancillary_files
+    return download_path
 
 
 def release_type_handler(query_params):
     """Handle 'release' type requests."""
-    start_date = datetime.datetime.strptime(query_params["start_date"], "%Y%m%d")
-    end_date = datetime.datetime.strptime(query_params["end_date"], "%Y%m%d")
-    exclude_file = query_params.get("exclude_file", None)
-    release_number = int(query_params["release_number"])
-    instrument = query_params["instrument"]
-
-    with db.Session() as session:
-        science_files_to_exclude = []
-        ancillary_files_to_exclude = []
-
-        if exclude_file:
-            science_files_to_exclude, ancillary_files_to_exclude = download_read_file(
-                exclude_file
-            )
-
-        science_files_to_update = query_latest_science_files(
-            session,
-            instrument,
-            start_date,
-            end_date,
-            science_files_to_exclude=science_files_to_exclude,
-        )
-
-        ancillary_files_to_update = get_latest_ancillary_files(
-            session,
-            instrument,
-            start_date,
-            end_date,
-            ancillary_files_to_exclude=ancillary_files_to_exclude,
-        )
-
-        if not science_files_to_update and not ancillary_files_to_update:
-            return {
-                "statusCode": 400,
-                "body": json.dumps(
-                    "No files to release for the specified "
-                    f"{instrument}, {start_date} to {end_date}, and "
-                    f"{release_number}."
-                ),
-            }
-
-        for obj in science_files_to_update:
-            obj.released = True
-
-        for obj in ancillary_files_to_update:
-            obj.released = True
-
-        session.commit()
+    manifest_file = query_params.get("manifest_file", None)
 
+    if manifest_file is None:
         return {
-            "statusCode": 200,
+            "statusCode": 400,
             "body": json.dumps(
-                f"Successfully released "
-                f"{len(science_files_to_update)} science files and "
-                f"{len(ancillary_files_to_update)} ancillary files."
+                "Missing required query parameter 'manifest_file' "
+                "for release operation."
             ),
         }
 
-
-def early_release_type_handler(query_params):
-    """Handle early-release requests using manifest file."""
-    manifest_file = query_params["manifest_file"]
-
-    science_files, ancillary_files = download_read_file(manifest_file)
-
-    if not science_files and not ancillary_files:
-        return {
-            "statusCode": 400,
-            "body": json.dumps("No files found in the manifest file."),
-        }
-
     with db.Session() as session:
-        if science_files:
-            session.query(models.ScienceFiles).filter(
-                models.ScienceFiles.file_path.in_(science_files)
-            ).update(
-                {models.ScienceFiles.released: True},
-                synchronize_session=False,
-            )
-
-        if ancillary_files:
-            session.query(models.AncillaryFiles).filter(
-                models.AncillaryFiles.file_path.in_(ancillary_files)
-            ).update(
-                {models.AncillaryFiles.released: True},
-                synchronize_session=False,
-            )
+        manifest_path = download_file(manifest_file)
+
+        manifest_file_obj = generate_imap_file_path(manifest_file)
+        start_date = datetime.datetime.strptime(manifest_file_obj.start_date, "%Y%m%d")
+        end_date = datetime.datetime.strptime(manifest_file_obj.end_date, "%Y%m%d")
+
+        # Read all lines in manifest file. We don't use numpy or pandas here
+        # because library will make lambda layer exceed its size limit.
+        all_lines = manifest_path.read_text(encoding="utf-8").splitlines()
+        for line in all_lines:
+            if (parsed_line := parse_manifest_line(line)) is None:
+                continue  # Skip comment, empty, or header lines
+
+            _, data_type, _, release_flag = parsed_line
+            # If row is to exclude, skip release process.
+            if not release_flag:
+                continue
+            logger.info(f"Releasing files for line: {line}")
+            if data_type == "all":
+                # Release all data for given instrument for the date
+                # range, including both ancillary and all data levels.
+                # Eg. hit, all, all, true
+                latest_science_release(
+                    session=session, start_date=start_date, end_date=end_date, line=line
+                )
+                latest_ancillary_release(session, start_date, end_date, line)
+            elif data_type == "ancillary":
+                latest_ancillary_release(session, start_date, end_date, line)
+            else:
+                latest_science_release(
+                    session=session, start_date=start_date, end_date=end_date, line=line
+                )
 
         session.commit()
 
-    logger.info(
-        f"Early released {len(science_files)} science files and "
-        f"{len(ancillary_files)} ancillary files."
-    )
-
-    return {
-        "statusCode": 200,
-        "body": json.dumps(
-            f"Successfully early released "
-            f"{len(science_files)} science files and "
-            f"{len(ancillary_files)} ancillary files."
-        ),
-    }
-
-
-def unrelease_type_handler(query_params):
-    """Handle unrelease requests using manifest file."""
-    manifest_file = query_params["manifest_file"]
-
-    science_files, ancillary_files = download_read_file(manifest_file)
-
-    if not science_files and not ancillary_files:
         return {
-            "statusCode": 400,
+            "statusCode": 200,
             "body": json.dumps(
-                "The manifest file does not contain any science or ancillary files."
+                f"Successfully released data per specification in {manifest_file}"
             ),
         }
 
-    with db.Session() as session:
-        if science_files:
-            session.query(models.ScienceFiles).filter(
-                models.ScienceFiles.file_path.in_(science_files)
-            ).update(
-                {models.ScienceFiles.released: False},
-                synchronize_session=False,
-            )
-
-        if ancillary_files:
-            session.query(models.AncillaryFiles).filter(
-                models.AncillaryFiles.file_path.in_(ancillary_files)
-            ).update(
-                {models.AncillaryFiles.released: False},
-                synchronize_session=False,
-            )
 
-        session.commit()
+def early_release_type_handler(query_params):
+    """Handle early-release requests using manifest file."""
+    return {"statusCode": 501, "body": "Early release operation not supported yet."}
 
-    logger.info(
-        f"Unreleased {len(science_files)} science files and "
-        f"{len(ancillary_files)} ancillary files."
-    )
 
-    return {
-        "statusCode": 200,
-        "body": json.dumps(
-            f"Successfully unreleased "
-            f"{len(science_files)} science files and "
-            f"{len(ancillary_files)} ancillary files."
-        ),
-    }
+def unrelease_type_handler(query_params):
+    """Handle unrelease requests using manifest file."""
+    return {"statusCode": 501, "body": "Unrelease operation not supported yet."}
 
 
 def reprocess_type_handler(query_params):
@@ -607,7 +406,7 @@ def reprocess_type_handler(query_params):
     NOTE: This may not be needed. If not needed, remove support
     at imap-data-access before removing this.
     """
-    return {"statusCode": 200, "body": "Reprocess for data release not supported yet."}
+    return {"statusCode": 501, "body": "Reprocess for data release not supported yet."}
 
 
 def lambda_handler(event, context):
diff --git a/sds_data_manager/lambda_code/SDSCode/api_lambdas/utils.py b/sds_data_manager/lambda_code/SDSCode/api_lambdas/utils.py
index d118dfe67..7fb0ef654 100644
--- a/sds_data_manager/lambda_code/SDSCode/api_lambdas/utils.py
+++ b/sds_data_manager/lambda_code/SDSCode/api_lambdas/utils.py
@@ -1,6 +1,11 @@
 """API utils."""
 
 import logging
+from collections.abc import Sequence
+
+from sqlalchemy import ColumnElement, Select, func, select
+
+from ..database.models import FILE_ID_COLUMNS, ScienceFiles
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
@@ -25,3 +30,49 @@ def is_authenticated_user(event):
         True if the path is authenticated, False otherwise
     """
     return event.get("rawPath", "").startswith(("/authorized", "/api-key"))
+
+
+def build_latest_version_query(
+    filters: Sequence[ColumnElement] = (),
+    major_only: bool = False,
+) -> Select:
+    """Build a query selecting the latest version of each science file.
+
+    Rows that share every :data:`FILE_ID_COLUMNS` value are just different
+    versions of the same file. The query uses a window function to rank them
+    by version and keeps only the ones with rank=1.
+
+    Parameters
+    ----------
+    filters : sequence of column expressions, optional
+        WHERE conditions applied *before* the version selection.
+    major_only : bool, optional
+        When True, return every minor version of each series' latest major
+        version instead of just the single newest file.
+
+    Returns
+    -------
+    Select
+        A SELECT of the :class:`ScienceFiles` table columns, restricted to the
+        latest version of each series.
+
+    """
+    table = ScienceFiles.__table__
+
+    partition_by = [table.c[column] for column in FILE_ID_COLUMNS]
+    order_by = [table.c.major_version.desc()]
+    if not major_only:
+        order_by.append(table.c.minor_version.desc())
+    rank = (
+        func.rank()
+        .over(partition_by=partition_by, order_by=order_by)
+        .label("version_rank")
+    )
+    ranked = select(table, rank).where(*filters).subquery()
+
+    rank_col = ranked.c[rank.name]
+    top_only = select(ranked).where(rank_col == 1)
+
+    # excludes the added RANK column
+    original_columns = [ranked.c[col.name] for col in table.c]
+    return top_only.with_only_columns(*original_columns)
diff --git a/sds_data_manager/lambda_code/SDSCode/database/models.py b/sds_data_manager/lambda_code/SDSCode/database/models.py
index 55d5c5b69..5824e5804 100644
--- a/sds_data_manager/lambda_code/SDSCode/database/models.py
+++ b/sds_data_manager/lambda_code/SDSCode/database/models.py
@@ -54,6 +54,15 @@
 # instrument's data. It's nice to have but not necessary.
 DEPENDENCY_RELATIONSHIPS = SqlEnum("SOFT", "HARD", name="dependency_relationship")
 
+# Columns identifying one logical CDF file; its rows differ only by version.
+FILE_ID_COLUMNS = (
+    "instrument",
+    "data_level",
+    "descriptor",
+    "start_date",
+    "repointing",
+)
+
 
 class Status(Enum):
     """Enum to store the status."""
@@ -181,11 +190,7 @@ def __table_args__(cls):  # noqa: N805
             Index(
                 # separate name for index for each subclass
                 f"idx_{cls.__tablename__}_version",
-                "instrument",
-                "data_level",
-                "descriptor",
-                "start_date",
-                "repointing",
+                *FILE_ID_COLUMNS,
                 "major_version",
                 "minor_version",
             ),
diff --git a/tests/lambda_endpoints/test_release_api.py b/tests/lambda_endpoints/test_release_api.py
index db8c7ddb0..81a2af482 100644
--- a/tests/lambda_endpoints/test_release_api.py
+++ b/tests/lambda_endpoints/test_release_api.py
@@ -83,20 +83,9 @@ def _ancillary(
     session.commit()
 
 
-# ---------------------------------------------------------------------------
-#   release — all files for instrument + date range + release number
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "sds_data_manager.lambda_code.SDSCode.api_lambdas.release_api.download_read_file"
-)
-def test_release_all_files_in_date_range(mock_download_read_file, session):
-    """release_type=release with no descriptor releases every matching file.
-
-    Two files for the target instrument within the date range and one file
-    outside the range must remain unreleased.
-    """
+@patch("sds_data_manager.lambda_code.SDSCode.api_lambdas.release_api.download_file")
+def test_science_release(mock_download_file, session, tmp_path):
+    """Test that science files in the manifest are released properly."""
     _science(
         session,
         file_path="imap/hit/l0/imap_hit_l0_hk_20250110_v001.0000.pkts",
@@ -126,19 +115,14 @@ def test_release_all_files_in_date_range(mock_download_read_file, session):
     )  # outside range
 
     # Provide the manifest file with the two in-range files
-    science_files = [
-        "imap/hit/l0/imap_hit_l0_hk_20250110_v001.0000.pkts",
-    ]
-    ancillary_files = []
-    mock_download_read_file.return_value = (science_files, ancillary_files)
+    file_content = """hit, l0, hk, false\nhit, l0, sci, true"""
+    manifest_path = tmp_path / "imap_hit_release_20250101_20250131_v001.txt"
+    manifest_path.write_text(file_content, encoding="utf-8")
+    mock_download_file.return_value = manifest_path
 
     params = {
-        "instrument": "hit",
-        "start_date": "20250101",
-        "end_date": "20250131",
         "release_type": "release",
-        "release_number": "1",
-        "exclude_file": "some_file.txt",
+        "manifest_file": "imap_hit_release_20250101_20250131_v001.txt",
     }
     result = release_api.lambda_handler(
         event=_build_event(params),
@@ -149,38 +133,18 @@ def test_release_all_files_in_date_range(mock_download_read_file, session):
 
     rows = {r.file_path: r.released for r in session.query(models.ScienceFiles).all()}
     assert rows["imap/hit/l0/imap_hit_l0_hk_20250110_v001.0000.pkts"] is False, (
-        "Excluded in-range file should not be released"
-    )
-    assert rows["imap/hit/l0/imap_hit_l0_sci_20250120_v001.0000.pkts"] is True, (
-        "Non-excluded in-range file should be released"
+        "HK descriptor file should remain unreleased"
     )
     assert rows["imap/hit/l0/imap_hit_l0_hk_20250201_v001.0000.pkts"] is False, (
-        "Out-of-range file must stay unreleased"
-    )
-
-    # Now test without exclude file.
-    params.pop("exclude_file")
-    result = release_api.lambda_handler(
-        event=_build_event(params),
-        context={},
-    )
-    assert result["statusCode"] == 200
-    rows = {r.file_path: r.released for r in session.query(models.ScienceFiles).all()}
-    assert rows["imap/hit/l0/imap_hit_l0_hk_20250110_v001.0000.pkts"] is True, (
-        "In-range file should be released"
+        "HK descriptor file should remain unreleased"
     )
     assert rows["imap/hit/l0/imap_hit_l0_sci_20250120_v001.0000.pkts"] is True, (
-        "In-range file should be released"
-    )
-    assert rows["imap/hit/l0/imap_hit_l0_hk_20250201_v001.0000.pkts"] is False, (
-        "Out-of-range file must stay unreleased"
+        "Sci descriptor file should be released"
     )
 
 
-# ---------------------------------------------------------------------------
-#  ancillary files release
-# ---------------------------------------------------------------------------
-def test_release_ancillary_files_in_date_range(session):
+@patch("sds_data_manager.lambda_code.SDSCode.api_lambdas.release_api.download_file")
+def test_ancillary_release(mock_download_file, session, tmp_path):
     """Ancillary files in manifest are released properly."""
     # Add all as unreleased
     session.add(
@@ -238,11 +202,11 @@ def test_release_ancillary_files_in_date_range(session):
     )
     session.commit()
 
-    result = release_api.get_latest_ancillary_files(
+    result = release_api.latest_ancillary_release(
         session,
-        instrument="codice",
         start_date=datetime.datetime.strptime("20260403", "%Y%m%d"),
         end_date=datetime.datetime.strptime("20260430", "%Y%m%d"),
+        line="codice,ancillary,l1a-sci-lut,true",
     )
     expected_ancillary_files = [
         "imap/ancillary/codice/imap_codice_l1a-sci-lut_20260403_20260403_v001.json",
@@ -254,41 +218,21 @@ def test_release_ancillary_files_in_date_range(session):
         f"got {[f.file_path for f in result]}"
     )
 
-
-# ---------------------------------------------------------------------------
-# ancillary release with exclude file
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "sds_data_manager.lambda_code.SDSCode.api_lambdas.release_api.download_read_file"
-)
-def test_ancillary_release_with_exclude_file(mock_download_read_file, session):
-    """release_type=release excludes specified ancillary files.
-
-    Three ancillary files for swe/l1b-in-flight-cal:
-      - v001 (older version, same descriptor+date) — should stay unreleased because
-        v002 is the latest.
-      - v002 (latest version, in range) — should be released because it is the
-        latest and is NOT in the exclude list.
-      - v001 with a different start_date inside range, but listed in the exclude
-        file — must remain unreleased.
-    """
-    # Older version — superseded by v002, same start_date
+    # Date edge cases
     _ancillary(
         session,
-        file_path="imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260413_v001.csv",
+        file_path="imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260401_20260413_v001.csv",
         version="v001",
-        start_date="20260413",
+        start_date="20260401",
+        end_date="20260413",
     )
-    # Latest version — should be released
     _ancillary(
         session,
-        file_path="imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260413_v002.csv",
+        file_path="imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260401_20260413_v002.csv",
         version="v002",
-        start_date="20260413",
+        start_date="20260401",
+        end_date="20260413",
     )
-    # In-range file that is the only version for its date but is in the exclude list
     _ancillary(
         session,
         file_path="imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260420_v001.csv",
@@ -302,54 +246,40 @@ def test_ancillary_release_with_exclude_file(mock_download_read_file, session):
         start_date="20260420",
     )
 
-    # Exclude list contains the April 20 file
-    mock_download_read_file.return_value = (
-        [],  # no science files excluded
-        ["imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260420_v002.csv"],
-    )
+    file_content = """swe, ancillary, l1b-in-flight-cal, true"""
+    manifest_path = tmp_path / "imap_swe_release_20260401_20260430_v001.txt"
+    manifest_path.write_text(file_content, encoding="utf-8")
+    mock_download_file.return_value = manifest_path
 
     params = {
-        "instrument": "swe",
-        "start_date": "20260401",
-        "end_date": "20260430",
         "release_type": "release",
-        "release_number": "1",
-        "exclude_file": "s3://dummy-bucket/exclude.txt",
+        "manifest_file": "s3://dummy-bucket/imap_swe_release_20260401_20260430_v001.txt",
     }
     result = release_api.lambda_handler(event=_build_event(params), context={})
 
     assert result["statusCode"] == 200
 
     rows = {r.file_path: r.released for r in session.query(models.AncillaryFiles).all()}
-    # Older version: not selected as latest → stays unreleased
     assert (
-        rows["imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260413_v001.csv"] is False
+        rows["imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260401_20260413_v001.csv"]
+        is False
     ), "Older version should not be released"
-    # Latest version, not excluded → should be released
+    # Latest version should be released
     assert (
-        rows["imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260413_v002.csv"] is True
+        rows["imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260401_20260413_v002.csv"]
+        is True
     ), "Latest version should be released"
-    # In-range but explicitly excluded → must stay unreleased
     assert (
         rows["imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260420_v001.csv"] is False
     ), "Excluded file should not be released"
-    # Latest version for April 20 -> must stay unreleased
+    # Latest version should be released
     assert (
-        rows["imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260420_v002.csv"] is False
-    ), "Latest version for April 20 should not be released"
-
-
-# ---------------------------------------------------------------------------
-#  ancillary release without exclude file — only latest version released
-# ---------------------------------------------------------------------------
+        rows["imap/ancillary/swe/imap_swe_l1b-in-flight-cal_20260420_v002.csv"] is True
+    ), "Latest version for April 20 should be released"
 
 
-@patch(
-    "sds_data_manager.lambda_code.SDSCode.api_lambdas.release_api.download_read_file"
-)
-def test_ancillary_release_without_exclude_file_latest_version_only(
-    mock_download_read_file, session
-):
+@patch("sds_data_manager.lambda_code.SDSCode.api_lambdas.release_api.download_file")
+def test_ancillary_release_with_wildcard(mock_download_file, session, tmp_path):
     """release_type=release with no exclude file releases only the latest version.
 
     Two versions of the same descriptor+start_date exist. Only the highest
@@ -378,13 +308,15 @@ def test_ancillary_release_without_exclude_file_latest_version_only(
         start_date="20260505",
     )
 
+    file_content = """swe, ancillary, all, true"""
+    manifest_path = tmp_path / "imap_swe_release_20260401_20260430_v001.txt"
+    manifest_path.write_text(file_content, encoding="utf-8")
+    mock_download_file.return_value = manifest_path
+
     # No exclude file provided
     params = {
-        "instrument": "swe",
-        "start_date": "20260401",
-        "end_date": "20260430",
         "release_type": "release",
-        "release_number": "1",
+        "manifest_file": "s3://dummy-bucket/imap_swe_release_20260401_20260430_v001.txt",
     }
     result = release_api.lambda_handler(event=_build_event(params), context={})
 
@@ -405,149 +337,37 @@ def test_ancillary_release_without_exclude_file_latest_version_only(
     ), "Out-of-range file must not be released"
 
 
-# ---------------------------------------------------------------------------
-#   early-release
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "sds_data_manager.lambda_code.SDSCode.api_lambdas.release_api.download_read_file"
-)
-def test_early_release(mock_download_read_file, session):
-    # Provide the manifest file with the two in-range files
-    science_files = [
-        "imap/hit/l0/imap_hit_l0_hk_20250110_v000.0001.pkts",
-        "imap/hit/l0/imap_hit_l0_sci_20250120_v000.0001.pkts",
-    ]
-    ancillary_files = []
-    mock_download_read_file.return_value = (science_files, ancillary_files)
-
-    _science(
-        session,
-        file_path="imap/hit/l0/imap_hit_l0_hk_20250110_v000.0001.pkts",
-        instrument="hit",
-        descriptor="hk",
-        start_date="20250110",
-        major_version=0,
-        minor_version=1,
-    )
-    _science(
-        session,
-        file_path="imap/hit/l0/imap_hit_l0_sci_20250120_v000.0001.pkts",
-        instrument="hit",
-        descriptor="sci",
-        start_date="20250120",
-        major_version=0,
-        minor_version=1,
-    )
-    _science(
-        session,
-        file_path="imap/hit/l0/imap_hit_l0_hk_20250201_v000.0001.pkts",
-        instrument="hit",
-        descriptor="hk",
-        start_date="20250201",
-        major_version=0,
-        minor_version=1,
-    )
-
-    manifest_file = "s3://dummy-bucket/manifest.txt"
+def test_early_release():
     result = release_api.lambda_handler(
         event=_build_event(
             {
                 "release_type": "early-release",
-                "manifest_file": manifest_file,
+                "manifest_file": "s3://dummy-bucket/manifest.txt",
             }
         ),
         context={},
     )
 
-    assert result["statusCode"] == 200
-
-    rows = {r.file_path: r.released for r in session.query(models.ScienceFiles).all()}
-    assert rows["imap/hit/l0/imap_hit_l0_hk_20250110_v000.0001.pkts"] is True
-    assert rows["imap/hit/l0/imap_hit_l0_sci_20250120_v000.0001.pkts"] is True
-    assert rows["imap/hit/l0/imap_hit_l0_hk_20250201_v000.0001.pkts"] is False
+    assert result["statusCode"] == 501
+    assert result["body"] == "Early release operation not supported yet."
 
 
-# ---------------------------------------------------------------------------
-#   unrelease
-# ---------------------------------------------------------------------------
-
-
-@patch(
-    "sds_data_manager.lambda_code.SDSCode.api_lambdas.release_api.download_read_file"
-)
-def test_unrelease_all_files_in_date_range(mock_download_read_file, session):
-    # Provide the manifest file with the two in-range files
-    science_files = [
-        "imap/hit/l0/imap_hit_l0_hk_20250110_v000.0001.pkts",
-        "imap/hit/l0/imap_hit_l0_sci_20250120_v000.0001.pkts",
-    ]
-    ancillary_files = []
-    mock_download_read_file.return_value = (science_files, ancillary_files)
-
-    _science(
-        session,
-        file_path="imap/hit/l0/imap_hit_l0_hk_20250110_v000.0001.pkts",
-        instrument="hit",
-        descriptor="hk",
-        start_date="20250110",
-        released=True,
-        major_version=0,
-        minor_version=1,
-    )
-    _science(
-        session,
-        file_path="imap/hit/l0/imap_hit_l0_sci_20250120_v000.0001.pkts",
-        instrument="hit",
-        descriptor="sci",
-        start_date="20250120",
-        released=True,
-        major_version=0,
-        minor_version=1,
-    )
-    _science(
-        session,
-        file_path="imap/hit/l0/imap_hit_l0_hk_20250201_v000.0001.pkts",
-        instrument="hit",
-        descriptor="hk",
-        start_date="20250201",
-        released=True,
-        major_version=0,
-        minor_version=1,
-    )  # outside range
-
-    manifest_file = "s3://dummy-bucket/manifest.txt"
+def test_unrelease_all_files_in_date_range():
     result = release_api.lambda_handler(
         event=_build_event(
             {
                 "release_type": "unrelease",
-                "manifest_file": manifest_file,
+                "manifest_file": "s3://dummy-bucket/manifest.txt",
             }
         ),
         context={},
     )
 
-    assert result["statusCode"] == 200
-
-    rows = {r.file_path: r.released for r in session.query(models.ScienceFiles).all()}
-    assert rows["imap/hit/l0/imap_hit_l0_hk_20250110_v000.0001.pkts"] is False, (
-        "In-range file must be unreleased"
-    )
-    assert rows["imap/hit/l0/imap_hit_l0_sci_20250120_v000.0001.pkts"] is False, (
-        "In-range file must be unreleased"
-    )
-    assert rows["imap/hit/l0/imap_hit_l0_hk_20250201_v000.0001.pkts"] is True, (
-        "Out-of-range file must remain released"
-    )
-
-
-# ---------------------------------------------------------------------------
-#  repoint files for a single day
-# ---------------------------------------------------------------------------
+    assert result["statusCode"] == 501
+    assert result["body"] == "Unrelease operation not supported yet."
 
 
-def test_release_repoint_files_date_range(session):
+def test_latest_science_release(session):
     """Test multiple repoint files for a single day."""
     # April 7th, Hi has three repoint files with different repointing and major
     # /minor versions
@@ -576,11 +396,11 @@ def test_release_repoint_files_date_range(session):
     session.commit()
 
     # Query for all files on this date
-    results = release_api.query_latest_science_files(
+    results = release_api.latest_science_release(
         session,
-        instrument="hi",
         start_date=datetime.datetime.strptime("20260407", "%Y%m%d"),
         end_date=datetime.datetime.strptime("20260407", "%Y%m%d"),
+        line="hi, l1a, all, true",
     )
     file_paths = sorted([obj.file_path for obj in results])
     assert file_paths == [
@@ -591,19 +411,21 @@ def test_release_repoint_files_date_range(session):
 
     # Query non-repoint files
     files = [
-        ("imap_swapi_l1_sci_20260407_v002.0002.cdf", "20260407", 2, 2),
-        ("imap_swapi_l1_sci_20260407_v001.0002.cdf", "20260407", 1, 2),
-        ("imap_swapi_l1_sci_20260407_v001.0001.cdf", "20260407", 1, 1),
-        ("imap_swapi_l1_sci_20260408_v001.0001.cdf", "20260408", 1, 1),
-        ("imap_swapi_l1_sci_20260408_v001.0002.cdf", "20260408", 1, 2),
+        ("imap_swapi_l1_sci_20260407_v002.0002.cdf", "sci", "20260407", 2, 2),
+        ("imap_swapi_l1_sci_20260407_v001.0002.cdf", "sci", "20260407", 1, 2),
+        ("imap_swapi_l1_sci_20260407_v001.0001.cdf", "sci", "20260407", 1, 1),
+        ("imap_swapi_l1_sci_20260408_v001.0001.cdf", "sci", "20260408", 1, 1),
+        ("imap_swapi_l1_sci_20260408_v001.0002.cdf", "sci", "20260408", 1, 2),
+        # HK is used to see if it gets excluded properly in later step
+        ("imap_swapi_l1a_hk_20260408_v001.0001.cdf", "hk", "20260408", 1, 1),
     ]
-    for file_path, start_date, major_ver, minor_ver in files:
+    for file_path, descriptor, start_date, major_ver, minor_ver in files:
         session.add(
             models.ScienceFiles(
                 file_path=file_path,
                 instrument="swapi",
                 data_level="l1",
-                descriptor="sci",
+                descriptor=descriptor,
                 start_date=datetime.datetime.strptime(start_date, "%Y%m%d"),
                 repointing=None,
                 major_version=major_ver,
@@ -615,22 +437,24 @@ def test_release_repoint_files_date_range(session):
         )
     session.commit()
 
-    latest_non_repoint_files = release_api.query_latest_science_files(
+    latest_non_repoint_files = release_api.latest_science_release(
         session,
-        instrument="swapi",
         start_date=datetime.datetime.strptime("20260407", "%Y%m%d"),
         end_date=datetime.datetime.strptime("20260407", "%Y%m%d"),
+        line="swapi,all,all,true",
     )
     file_paths = sorted([obj.file_path for obj in latest_non_repoint_files])
     assert file_paths == [
         "imap_swapi_l1_sci_20260407_v002.0002.cdf",
     ], f"Expected only the latest non-repoint file, got: {file_paths}"
 
-    latest_non_repoint_files = release_api.query_latest_science_files(
+    # In this release query, we only ask for latest sci files on April 8th,
+    # so the HK file should be excluded and should not be returned.
+    latest_non_repoint_files = release_api.latest_science_release(
         session,
-        instrument="swapi",
         start_date=datetime.datetime.strptime("20260408", "%Y%m%d"),
         end_date=datetime.datetime.strptime("20260408", "%Y%m%d"),
+        line="swapi,l1,sci,true",
     )
     file_paths = sorted([obj.file_path for obj in latest_non_repoint_files])
     assert file_paths == [