Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "timdex_dataset_api"
version = "5.1.0"
version = "5.2.0"
description = "Python library for interacting with a TIMDEX parquet dataset"
readme = "README.md"
requires-python = ">=3.12"
Expand Down
35 changes: 34 additions & 1 deletion tests/test_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -262,7 +262,6 @@ def test_tdm_append_deltas_view_empty_structure(timdex_metadata):
"run_record_offset",
"run_timestamp",
"filename",
"append_delta_filename",
}
assert set(append_deltas_df.columns) == expected_columns
assert len(append_deltas_df) == 0
Expand Down Expand Up @@ -290,6 +289,10 @@ def test_tdm_append_deltas_count_property_empty(timdex_metadata):
assert timdex_metadata.append_deltas_count == 0


def test_tdm_append_deltas_count_for_missing_view(timdex_metadata_empty):
assert timdex_metadata_empty.append_deltas_count_for(TIMDEXRecords) == 0


def test_tdm_records_equals_static_without_deltas(timdex_metadata):
static_count = timdex_metadata.timdex_dataset.conn.query(
"""select count(*) from static_db.records;"""
Expand Down Expand Up @@ -323,6 +326,30 @@ def test_tdm_append_deltas_view_has_data(timdex_metadata_with_deltas):
assert append_deltas_count > 0


def test_tdm_filename_filter_matches_append_delta_rows(timdex_metadata_with_deltas):
conn = timdex_metadata_with_deltas.timdex_dataset.conn
filename = conn.execute(
"select filename from metadata.records_append_deltas limit 1"
).fetchone()[0]

delta_count = conn.execute(
"select count(*) from metadata.records_append_deltas where filename = ?",
[filename],
).fetchone()[0]
records_count = conn.execute(
"select count(*) from metadata.records where filename = ?", [filename]
).fetchone()[0]
shadow_count = conn.execute("""
select count(*)
from metadata.records_append_deltas
where filename like '%append_delta%'
""").fetchone()[0]
Comment thread
ghukill marked this conversation as resolved.

assert delta_count > 0
assert records_count >= delta_count
assert shadow_count == 0


def test_tdm_records_includes_deltas(timdex_metadata_with_deltas):
static_count = timdex_metadata_with_deltas.timdex_dataset.conn.query(
"""select count(*) from static_db.records;"""
Expand Down Expand Up @@ -434,6 +461,12 @@ def test_tdm_merge_append_deltas_deletes_append_deltas(
assert not os.listdir(records_deltas_path_after)


def test_tdm_append_deltas_count_for_after_merge(timdex_metadata_with_deltas):
timdex_metadata_with_deltas.merge_append_deltas()

assert timdex_metadata_with_deltas.append_deltas_count_for(TIMDEXRecords) == 0


def test_tdm_embeddings_metadata_view_structure(tmp_path):
td = TIMDEXDataset(str(tmp_path / "embeddings_metadata_structure"))

Expand Down
31 changes: 12 additions & 19 deletions timdex_dataset_api/metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -129,9 +129,12 @@ def current_records_count(self) -> int:
def append_deltas_count_for(self, data_type_class: type["TIMDEXDataType"]) -> int:
"""Count append deltas rows for a single data type."""
view_name = f"{data_type_class.NAME}_append_deltas"
return self.timdex_dataset.conn.query(f"""
select count(*) from metadata.{view_name};
""").fetchone()[0] # type: ignore[index]
try:
return self.timdex_dataset.conn.query(f"""
select count(*) from metadata.{view_name};
""").fetchone()[0] # type: ignore[index]
except (DuckDBCatalogException, DuckDBIOException):
Comment thread
ghukill marked this conversation as resolved.
return 0
Comment thread
ghukill marked this conversation as resolved.

@property
def append_deltas_count(self) -> int:
Expand Down Expand Up @@ -347,10 +350,7 @@ def _create_append_deltas_view(
conn.execute(f"""
create or replace view metadata.{view_name} as (
select *
from read_parquet(
'{deltas_path}/*.parquet',
filename = 'append_delta_filename'
)
from read_parquet('{deltas_path}/*.parquet')
);
""")
return
Expand All @@ -365,8 +365,7 @@ def _create_append_deltas_view(
if table_exists:
conn.execute(f"""
create or replace view metadata.{view_name} as (
select *,
null::varchar as append_delta_filename
select *
from {static_table}
where 1 = 0
);
Expand Down Expand Up @@ -578,22 +577,16 @@ def merge_append_deltas(self) -> None:
all_delta_filenames: dict[str, list[str]] = {}
has_any_deltas = False
for data_type_class in self.data_type_classes:
deltas_view = f"{data_type_class.NAME}_append_deltas"
deltas_glob = f"{self.append_deltas_path_for(data_type_class)}/*.parquet"
Comment thread
ghukill marked this conversation as resolved.
try:
filenames = (
self.timdex_dataset.conn.query(f"""
select distinct(append_delta_filename)
from metadata.{deltas_view}
select file from glob('{deltas_glob}')
""")
.to_df()["append_delta_filename"]
.to_df()["file"]
.to_list()
)
except (
DuckDBIOException,
DuckDBCatalogException,
DuckDBBinderException,
KeyError,
):
except (DuckDBIOException, KeyError):
filenames = []
all_delta_filenames[data_type_class.NAME] = filenames
if filenames:
Expand Down
2 changes: 1 addition & 1 deletion uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading