diff --git a/.bumpversion.cfg b/.bumpversion.cfg
index a7be11f..3a8e09b 100644
--- a/.bumpversion.cfg
+++ b/.bumpversion.cfg
@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.0.7
+current_version = 0.0.8
 commit = True
 tag = True
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)
diff --git a/.github/workflows/workflow.yaml b/.github/workflows/workflow.yaml
index d08c2da..7e604e1 100644
--- a/.github/workflows/workflow.yaml
+++ b/.github/workflows/workflow.yaml
@@ -142,10 +142,10 @@ jobs:
           pattern: coverage-data-*
           merge-multiple: true
 
-      - name: Combine coverage and fail if it's <85%
+      - name: Combine coverage and fail if it's <100%
         run: |
           coverage combine
-          coverage report --fail-under=85
+          coverage report --fail-under=100
 
       - name: Upload HTML report
         if: ${{ failure() }}
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1bedbf9..74791dd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.0.8] - 2026-06-25
+
+### Added
+
+- Database-free unit tests covering the remaining branch and error paths
+  across the compression, hypercore, continuous aggregate, retention, and
+  hypertable subpackages, bringing total test coverage to 100%.
+
+### Changed
+
+- Raised the CI coverage gate from 85% to 100%.
+
 ## [0.0.7] - 2026-06-25
 
 Production-readiness hardening release.
diff --git a/pyproject.toml b/pyproject.toml
index 8ee7bc2..3a3e732 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "timescaledb"
-version = "0.0.7"
+version = "0.0.8"
 description = "TimescaleDB is a Python Client based on SQLModel and SQLAlchemy for high-performance real-time analytics time-series data."
 readme = "README.md"
 authors = [
diff --git a/src/timescaledb/__init__.py b/src/timescaledb/__init__.py
index 92d2b86..54b0a2c 100644
--- a/src/timescaledb/__init__.py
+++ b/src/timescaledb/__init__.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-__version__ = "0.0.7"
+__version__ = "0.0.8"
 
 from . import defaults, metadata
 from .activator import activate_timescaledb_extension
diff --git a/tests/test_unit_compression.py b/tests/test_unit_compression.py
new file mode 100644
index 0000000..c7dd75b
--- /dev/null
+++ b/tests/test_unit_compression.py
@@ -0,0 +1,286 @@
+"""Pure-unit tests for the compression subpackage.
+
+These exercise the SQL builders, extractors, validators, and the
+enable/add/sync helpers using lightweight fake sessions so no live database
+is required.
+"""
+
+from datetime import datetime
+from typing import Optional
+
+import pytest
+import sqlmodel
+from sqlalchemy import JSON, Column, MetaData
+from sqlalchemy.exc import SQLAlchemyError
+from sqlmodel import Field
+
+from timescaledb.compression import enable, extractors, sql, validators
+from timescaledb.compression.add import add_compression_policy
+from timescaledb.compression.enable import enable_table_compression
+from timescaledb.compression.sync import sync_compression_policies
+from timescaledb.exceptions import (
+    InvalidCompressionFields,
+    InvalidOrderByField,
+    InvalidSegmentByField,
+)
+
+from .conftest import Metric, SimpleCompression, VideoView
+
+
+class QueryRecorder:
+    def __init__(self):
+        self.queries = []
+        self.commits = 0
+        self.rollbacks = 0
+
+    def execute(self, query):
+        self.queries.append(str(query))
+
+    def commit(self):
+        self.commits += 1
+
+    def rollback(self):
+        self.rollbacks += 1
+
+
+class FailingSession(QueryRecorder):
+    def execute(self, query):
+        raise SQLAlchemyError("boom")
+
+
+# ---------------------------------------------------------------------------
+# Isolated model with a JSON column for validator type checks
+# ---------------------------------------------------------------------------
+
+
+class _CompBase(sqlmodel.SQLModel):
+    metadata = MetaData()
+
+
+class ValidatorModel(_CompBase, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    time: datetime = Field(sa_type=sqlmodel.DateTime)
+    value: int
+    name: str
+    payload: Optional[dict] = Field(default=None, sa_column=Column(JSON))
+
+
+# ---------------------------------------------------------------------------
+# add_compression_policy
+# ---------------------------------------------------------------------------
+
+
+def test_add_compression_policy_requires_model_or_table_name():
+    with pytest.raises(ValueError, match="model or table_name is required"):
+        add_compression_policy(QueryRecorder(), model=None, table_name=None)
+
+
+def test_add_compression_policy_skips_disabled_model():
+    session = QueryRecorder()
+    add_compression_policy(session, model=Metric)
+    assert session.queries == []
+    assert session.commits == 0
+
+
+def test_add_compression_policy_with_model_and_compress_after():
+    session = QueryRecorder()
+    add_compression_policy(session, model=SimpleCompression, compress_after="7 days")
+    assert session.commits == 1
+    assert "add_compression_policy" in session.queries[0]
+    assert "compress_after" in session.queries[0]
+
+
+def test_add_compression_policy_with_model_and_created_before():
+    session = QueryRecorder()
+    add_compression_policy(
+        session,
+        model=SimpleCompression,
+        compress_created_before="INTERVAL 7 days",
+    )
+    assert session.commits == 1
+    assert "compress_created_before" in session.queries[0]
+
+
+# ---------------------------------------------------------------------------
+# enable_table_compression
+# ---------------------------------------------------------------------------
+
+
+def test_enable_table_compression_requires_model_or_table_name():
+    with pytest.raises(ValueError, match="model or table_name is required"):
+        enable_table_compression(QueryRecorder(), model=None, table_name=None)
+
+
+def test_enable_table_compression_disabled_model_is_noop():
+    session = QueryRecorder()
+    enable_table_compression(session, model=Metric)
+    assert session.queries == []
+    assert session.commits == 0
+
+
+def test_enable_table_compression_returns_when_params_not_enabled(monkeypatch):
+    """Defensive branch: params present but compression disabled."""
+    monkeypatch.setattr(
+        enable.extractors,
+        "extract_model_compression_params",
+        lambda model: {"compress_enabled": False},
+    )
+    session = QueryRecorder()
+    enable_table_compression(session, model=SimpleCompression)
+    assert session.queries == []
+
+
+def test_enable_table_compression_model_without_orderby_segmentby():
+    session = QueryRecorder()
+    enable_table_compression(session, model=SimpleCompression)
+    assert session.commits == 1
+    assert "timescaledb.compress" in session.queries[0]
+
+
+def test_enable_table_compression_model_with_orderby_and_segmentby():
+    session = QueryRecorder()
+    enable_table_compression(session, model=VideoView)
+    assert session.commits == 1
+    assert "compress_orderby" in session.queries[0]
+    assert "compress_segmentby" in session.queries[0]
+
+
+def test_enable_table_compression_explicit_orderby_and_segmentby():
+    """Explicit args are not overwritten by model metadata."""
+    session = QueryRecorder()
+    enable_table_compression(
+        session,
+        model=SimpleCompression,
+        compress_orderby="value ASC",
+        compress_segmentby="value",
+    )
+    assert session.commits == 1
+    assert "value ASC" in session.queries[0]
+
+
+# ---------------------------------------------------------------------------
+# sql.format_compression_policy_sql_query
+# ---------------------------------------------------------------------------
+
+
+def test_format_compression_policy_returns_none_when_no_age():
+    assert sql.format_compression_policy_sql_query("metrics") is None
+
+
+def test_format_compression_policy_created_before_requires_timedelta():
+    with pytest.raises(ValueError, match="timedelta"):
+        sql.format_compression_policy_sql_query("metrics", compress_created_before=5)
+
+
+def test_format_compression_policy_after_invalid_template(monkeypatch):
+    monkeypatch.setattr(sql, "COMPRESSION_POLICY_SQL_TEMPLATE_MAPPING", {})
+    with pytest.raises(ValueError, match="Invalid interval type"):
+        sql.format_compression_policy_sql_query("metrics", compress_after="7 days")
+
+
+def test_format_compression_policy_before_invalid_template(monkeypatch):
+    monkeypatch.setattr(sql, "COMPRESSION_POLICY_SQL_TEMPLATE_MAPPING", {})
+    with pytest.raises(ValueError, match="Invalid interval type"):
+        sql.format_compression_policy_sql_query(
+            "metrics", compress_created_before="7 days"
+        )
+
+
+# ---------------------------------------------------------------------------
+# sync_compression_policies
+# ---------------------------------------------------------------------------
+
+
+def test_sync_compression_policies_with_explicit_models():
+    session = QueryRecorder()
+    sync_compression_policies(session, SimpleCompression)
+    assert session.commits == 1
+    assert any("timescaledb.compress" in q for q in session.queries)
+
+
+def test_sync_compression_policies_rolls_back_on_error():
+    session = FailingSession()
+    with pytest.raises(SQLAlchemyError):
+        sync_compression_policies(session, SimpleCompression)
+    assert session.rollbacks == 1
+
+
+# ---------------------------------------------------------------------------
+# extractors
+# ---------------------------------------------------------------------------
+
+
+def test_extract_model_compression_params_returns_none_when_disabled():
+    assert extractors.extract_model_compression_params(Metric) is None
+
+
+# ---------------------------------------------------------------------------
+# validators
+# ---------------------------------------------------------------------------
+
+
+def test_validate_segmentby_missing_field():
+    with pytest.raises(InvalidSegmentByField, match="not found"):
+        validators.validate_compress_segmentby_field(ValidatorModel, "missing")
+
+
+def test_validate_segmentby_invalid_type():
+    with pytest.raises(InvalidSegmentByField, match="invalid type"):
+        validators.validate_compress_segmentby_field(ValidatorModel, "payload")
+
+
+def test_validate_segmentby_none_is_valid():
+    assert validators.validate_compress_segmentby_field(ValidatorModel, None) is True
+
+
+def test_validate_orderby_empty_spec():
+    with pytest.raises(InvalidOrderByField, match="Empty orderby"):
+        validators.validate_compress_orderby_field(ValidatorModel, ",")
+
+
+def test_validate_orderby_missing_field():
+    with pytest.raises(InvalidOrderByField, match="not found"):
+        validators.validate_compress_orderby_field(ValidatorModel, "missing")
+
+
+def test_validate_orderby_invalid_type():
+    with pytest.raises(InvalidOrderByField, match="invalid type"):
+        validators.validate_compress_orderby_field(ValidatorModel, "payload")
+
+
+def test_validate_orderby_multiple_fields_without_direction():
+    assert (
+        validators.validate_compress_orderby_field(ValidatorModel, "value, name")
+        is True
+    )
+
+
+def test_validate_orderby_invalid_direction():
+    with pytest.raises(InvalidOrderByField, match="Invalid direction"):
+        validators.validate_compress_orderby_field(ValidatorModel, "value SIDEWAYS")
+
+
+def test_validate_orderby_incomplete_nulls_spec():
+    with pytest.raises(InvalidOrderByField, match="NULLS"):
+        validators.validate_compress_orderby_field(ValidatorModel, "value ASC NULLS")
+
+
+def test_validate_orderby_invalid_nulls_keyword():
+    with pytest.raises(InvalidOrderByField, match="NULLS"):
+        validators.validate_compress_orderby_field(
+            ValidatorModel, "value ASC FOO BAR"
+        )
+
+
+def test_validate_orderby_valid_nulls_spec():
+    assert (
+        validators.validate_compress_orderby_field(ValidatorModel, "value ASC NULLS FIRST")
+        is True
+    )
+
+
+def test_validate_unique_segmentby_and_orderby_conflict():
+    with pytest.raises(InvalidCompressionFields, match="must be different"):
+        validators.validate_unique_segmentby_and_orderby_fields(
+            ValidatorModel, segmentby_field="value", orderby_field="value"
+        )
diff --git a/tests/test_unit_continuous_aggregates.py b/tests/test_unit_continuous_aggregates.py
new file mode 100644
index 0000000..5a3210d
--- /dev/null
+++ b/tests/test_unit_continuous_aggregates.py
@@ -0,0 +1,150 @@
+"""Pure-unit tests covering continuous_aggregates edge cases and helpers."""
+
+from datetime import timedelta
+
+import pytest
+
+from timescaledb.continuous_aggregates import sql
+from timescaledb.continuous_aggregates.alter import add_generated_aggregate_column
+from timescaledb.continuous_aggregates.create import create_continuous_aggregate
+from timescaledb.continuous_aggregates.policies import (
+    add_continuous_aggregate_policy,
+    remove_continuous_aggregate_policy,
+)
+from timescaledb.continuous_aggregates.refresh import refresh_continuous_aggregate
+
+
+class QueryRecorder:
+    def __init__(self):
+        self.queries = []
+        self.commits = 0
+
+    def execute(self, query):
+        self.queries.append(str(query))
+
+    def commit(self):
+        self.commits += 1
+
+
+# ---------------------------------------------------------------------------
+# commit=False branches on the executing helpers
+# ---------------------------------------------------------------------------
+
+
+def test_create_continuous_aggregate_without_commit():
+    session = QueryRecorder()
+    create_continuous_aggregate(
+        session,
+        "v",
+        "SELECT 1 AS bucket",
+        with_data=False,
+        commit=False,
+    )
+    assert session.commits == 0
+    assert len(session.queries) == 1
+
+
+def test_refresh_continuous_aggregate_without_commit():
+    session = QueryRecorder()
+    refresh_continuous_aggregate(session, "v", commit=False)
+    assert session.commits == 0
+    assert len(session.queries) == 1
+
+
+def test_add_continuous_aggregate_policy_without_commit():
+    session = QueryRecorder()
+    add_continuous_aggregate_policy(
+        session,
+        "v",
+        start_offset="1 month",
+        end_offset="1 hour",
+        schedule_interval="1 hour",
+        commit=False,
+    )
+    assert session.commits == 0
+
+
+def test_remove_continuous_aggregate_policy_without_commit():
+    session = QueryRecorder()
+    remove_continuous_aggregate_policy(session, "v", commit=False)
+    assert session.commits == 0
+
+
+def test_add_generated_aggregate_column_without_commit():
+    session = QueryRecorder()
+    add_generated_aggregate_column(
+        session, "v", "max_temp", "DOUBLE PRECISION", "max(temp)", commit=False
+    )
+    assert session.commits == 0
+
+
+# ---------------------------------------------------------------------------
+# sql private helpers
+# ---------------------------------------------------------------------------
+
+
+def test_compile_sql_without_params():
+    assert "SELECT 1" in sql._compile_sql("SELECT 1")
+
+
+def test_clean_interval_fractional_timedelta():
+    assert sql._clean_interval(timedelta(milliseconds=1500)) == "1.5 seconds"
+
+
+def test_quote_qualified_identifier_requires_value():
+    with pytest.raises(ValueError, match="identifier is required"):
+        sql._quote_qualified_identifier("")
+
+
+def test_format_option_value_integer():
+    assert sql._format_option_value(5) == "5"
+
+
+def test_format_option_value_timedelta():
+    assert sql._format_option_value(timedelta(hours=1)) == "'3600 seconds'"
+
+
+# ---------------------------------------------------------------------------
+# sql public builders edge cases
+# ---------------------------------------------------------------------------
+
+
+def test_format_refresh_with_null_window():
+    out = sql.format_refresh_continuous_aggregate_sql_query("conditions")
+    assert "NULL, NULL" in out
+
+
+def test_format_create_continuous_aggregate_requires_select_query():
+    with pytest.raises(ValueError, match="select_query is required"):
+        sql.format_create_continuous_aggregate_sql("v", "")
+
+
+def test_format_add_generated_aggregate_column_requires_data_type():
+    with pytest.raises(ValueError, match="data_type is required"):
+        sql.format_add_generated_aggregate_column_sql("v", "c", "", "max(x)")
+
+
+def test_format_add_generated_aggregate_column_requires_expression():
+    with pytest.raises(ValueError, match="aggregate_expression is required"):
+        sql.format_add_generated_aggregate_column_sql("v", "c", "INT", "")
+
+
+def test_format_add_policy_requires_schedule_interval():
+    with pytest.raises(ValueError, match="schedule_interval is required"):
+        sql.format_add_continuous_aggregate_policy_sql_query(
+            "v",
+            start_offset="1 month",
+            end_offset="1 hour",
+            schedule_interval=None,
+        )
+
+
+def test_format_add_policy_with_null_offsets():
+    out = sql.format_add_continuous_aggregate_policy_sql_query(
+        "v",
+        start_offset=None,
+        end_offset=None,
+        schedule_interval="1 hour",
+    )
+    assert "start_offset => NULL" in out
+    assert "end_offset => NULL" in out
diff --git a/tests/test_unit_hypercore.py b/tests/test_unit_hypercore.py
new file mode 100644
index 0000000..3b484bb
--- /dev/null
+++ b/tests/test_unit_hypercore.py
@@ -0,0 +1,296 @@
+"""Pure-unit tests for the hypercore (columnstore) subpackage."""
+
+from datetime import datetime, timedelta
+
+import pytest
+import sqlmodel
+from sqlalchemy import MetaData
+from sqlalchemy.exc import SQLAlchemyError
+from sqlmodel import Field
+
+from timescaledb.hypercore import extractors, sql
+from timescaledb.hypercore.add import add_columnstore_policy
+from timescaledb.hypercore.convert import (
+    convert_to_columnstore,
+    convert_to_rowstore,
+)
+from timescaledb.hypercore.enable import enable_columnstore
+from timescaledb.hypercore.list import list_columnstore_policies
+from timescaledb.hypercore.remove import remove_columnstore_policy
+from timescaledb.hypercore.sync import sync_columnstore_policies
+
+from .conftest import Metric
+
+
+class IsolatedSQLModel(sqlmodel.SQLModel):
+    metadata = MetaData()
+
+
+class ColumnstoreMetric(IsolatedSQLModel, table=True):
+    id: int = Field(primary_key=True)
+    time: int = Field(primary_key=True)
+    device_id: int = Field(index=True)
+    value: float
+
+    __enable_columnstore__ = True
+    __columnstore_orderby__ = "time DESC"
+    __columnstore_segmentby__ = "device_id"
+    __columnstore_after__ = "7 days"
+    __columnstore_if_not_exists__ = True
+
+
+class ColumnstoreNoAge(IsolatedSQLModel, table=True):
+    id: int = Field(primary_key=True)
+    time: int = Field(primary_key=True)
+    value: float
+
+    __enable_columnstore__ = True
+
+
+class QueryRecorder:
+    def __init__(self):
+        self.queries = []
+        self.commits = 0
+        self.rollbacks = 0
+
+    def execute(self, query):
+        self.queries.append(str(query))
+
+    def commit(self):
+        self.commits += 1
+
+    def rollback(self):
+        self.rollbacks += 1
+
+
+class FailingSession(QueryRecorder):
+    def execute(self, query):
+        raise SQLAlchemyError("boom")
+
+
+class FetchSession:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def execute(self, query):
+        return self
+
+    def fetchall(self):
+        return self._rows
+
+
+# ---------------------------------------------------------------------------
+# add_columnstore_policy
+# ---------------------------------------------------------------------------
+
+
+def test_add_columnstore_policy_requires_model_or_table_name():
+    with pytest.raises(ValueError, match="model or table_name is required"):
+        add_columnstore_policy(QueryRecorder(), model=None, table_name=None)
+
+
+def test_add_columnstore_policy_disabled_model_is_noop():
+    session = QueryRecorder()
+    add_columnstore_policy(session, model=Metric)
+    assert session.queries == []
+
+
+def test_add_columnstore_policy_enabled_model_without_age_is_noop():
+    session = QueryRecorder()
+    add_columnstore_policy(session, model=ColumnstoreNoAge)
+    assert session.queries == []
+
+
+def test_add_columnstore_policy_with_table_name():
+    session = QueryRecorder()
+    add_columnstore_policy(session, table_name="metrics", after="7 days")
+    assert session.commits == 1
+    assert "add_columnstore_policy" in session.queries[0]
+
+
+# ---------------------------------------------------------------------------
+# enable_columnstore
+# ---------------------------------------------------------------------------
+
+
+def test_enable_columnstore_requires_model_or_table_name():
+    with pytest.raises(ValueError, match="model or table_name is required"):
+        enable_columnstore(QueryRecorder(), model=None, table_name=None)
+
+
+def test_enable_columnstore_disabled_model_is_noop():
+    session = QueryRecorder()
+    enable_columnstore(session, model=Metric)
+    assert session.queries == []
+
+
+def test_enable_columnstore_model_fills_orderby_segmentby():
+    session = QueryRecorder()
+    enable_columnstore(session, model=ColumnstoreMetric)
+    assert session.commits == 1
+    assert "timescaledb.orderby" in session.queries[0]
+    assert "timescaledb.segmentby" in session.queries[0]
+
+
+def test_enable_columnstore_model_with_explicit_orderby_segmentby():
+    session = QueryRecorder()
+    enable_columnstore(
+        session, model=ColumnstoreMetric, orderby="value DESC", segmentby="value"
+    )
+    assert session.commits == 1
+    assert "value DESC" in session.queries[0]
+
+
+def test_enable_columnstore_with_table_name():
+    session = QueryRecorder()
+    enable_columnstore(
+        session, table_name="metrics", orderby="time DESC", segmentby="device_id"
+    )
+    assert session.commits == 1
+    assert "timescaledb.enable_columnstore = true" in session.queries[0]
+
+
+# ---------------------------------------------------------------------------
+# convert helpers
+# ---------------------------------------------------------------------------
+
+
+def test_convert_to_columnstore_commits():
+    session = QueryRecorder()
+    convert_to_columnstore(session, "_chunk")
+    assert session.commits == 1
+    assert "convert_to_columnstore" in session.queries[0]
+
+
+def test_convert_to_columnstore_without_commit():
+    session = QueryRecorder()
+    convert_to_columnstore(session, "_chunk", commit=False)
+    assert session.commits == 0
+    assert len(session.queries) == 1
+
+
+def test_convert_to_rowstore_commits():
+    session = QueryRecorder()
+    convert_to_rowstore(session, "_chunk")
+    assert session.commits == 1
+    assert "convert_to_rowstore" in session.queries[0]
+
+
+def test_convert_to_rowstore_without_commit():
+    session = QueryRecorder()
+    convert_to_rowstore(session, "_chunk", commit=False)
+    assert session.commits == 0
+    assert len(session.queries) == 1
+
+
+# ---------------------------------------------------------------------------
+# list / remove
+# ---------------------------------------------------------------------------
+
+
+def test_list_columnstore_policies():
+    session = FetchSession([("metrics",), ("views",)])
+    assert list_columnstore_policies(session) == ["metrics", "views"]
+
+
+def test_remove_columnstore_policy_commits():
+    session = QueryRecorder()
+    remove_columnstore_policy(session, "metrics")
+    assert session.commits == 1
+    assert "remove_columnstore_policy" in session.queries[0]
+
+
+def test_remove_columnstore_policy_without_commit():
+    session = QueryRecorder()
+    remove_columnstore_policy(session, "metrics", commit=False)
+    assert session.commits == 0
+
+
+# ---------------------------------------------------------------------------
+# sync_columnstore_policies
+# ---------------------------------------------------------------------------
+
+
+def test_sync_columnstore_policies_rolls_back_on_error():
+    session = FailingSession()
+    with pytest.raises(SQLAlchemyError):
+        sync_columnstore_policies(session, ColumnstoreMetric)
+    assert session.rollbacks == 1
+
+
+# ---------------------------------------------------------------------------
+# extractors
+# ---------------------------------------------------------------------------
+
+
+def test_extract_columnstore_params_returns_none_when_disabled():
+    assert extractors.extract_model_columnstore_params(Metric) is None
+
+
+def test_extract_columnstore_policy_params_returns_none_when_disabled():
+    assert extractors.extract_model_columnstore_policy_params(Metric) is None
+
+
+def test_extract_columnstore_params_without_orderby_segmentby():
+    params = extractors.extract_model_columnstore_params(ColumnstoreNoAge)
+    assert params["columnstore_enabled"] is True
+    assert "orderby" not in params
+    assert "segmentby" not in params
+
+
+# ---------------------------------------------------------------------------
+# sql helpers
+# ---------------------------------------------------------------------------
+
+
+def test_quote_qualified_identifier_requires_value():
+    with pytest.raises(ValueError, match="identifier is required"):
+        sql.quote_qualified_identifier("")
+
+
+def test_clean_interval_value_invalid_type():
+    assert sql._clean_interval_value(1.5) == (1.5, "INVALID")
+
+
+def test_clean_interval_value_fractional_timedelta():
+    assert sql._clean_interval_value(timedelta(milliseconds=1500)) == (
+        "1.5 seconds",
+        "INTERVAL",
+    )
+
+
+def test_policy_interval_sql_invalid_type_raises():
+    with pytest.raises(ValueError, match="Invalid interval type"):
+        sql._policy_interval_sql("after", 1.5, allow_integer=True)
+
+
+def test_policy_interval_sql_integer_not_allowed_raises():
+    with pytest.raises(ValueError, match="Invalid interval type"):
+        sql._policy_interval_sql("created_before", 5, allow_integer=False)
+
+
+def test_compile_sql_without_params():
+    out = sql._compile_sql("SELECT 1")
+    assert "SELECT 1" in out
+
+
+def test_format_enable_columnstore_sql_without_orderby_segmentby():
+    out = sql.format_enable_columnstore_sql("metrics")
+    assert "timescaledb.enable_columnstore = true" in out
+    assert "timescaledb.orderby" not in out
+    assert "timescaledb.segmentby" not in out
+
+
+def test_format_add_columnstore_policy_with_all_options():
+    out = sql.format_add_columnstore_policy_sql_query(
+        "metrics",
+        created_before="3 months",
+        schedule_interval="1 hour",
+        initial_start=datetime(2024, 1, 1),
+        timezone="UTC",
+        if_not_exists=True,
+    )
+    assert "created_before => CAST('3 months' AS INTERVAL)" in out
+    assert "schedule_interval => CAST('1 hour' AS INTERVAL)" in out
+    assert "initial_start =>" in out
+    assert "timezone => 'UTC'" in out
diff --git a/tests/test_unit_misc.py b/tests/test_unit_misc.py
new file mode 100644
index 0000000..561f97b
--- /dev/null
+++ b/tests/test_unit_misc.py
@@ -0,0 +1,337 @@
+"""Pure-unit tests for defaults, retention, hypertable, hyperfunction, and
+query edge cases that close the remaining coverage gaps."""
+
+from datetime import datetime
+from typing import Optional
+
+import pytest
+import sqlmodel
+from sqlalchemy import MetaData
+from sqlalchemy.exc import SQLAlchemyError
+from sqlmodel import Field
+
+from timescaledb.defaults import get_defaults
+from timescaledb.exceptions import InvalidChunkTimeInterval
+from timescaledb.hyperfunctions.main import time_bucket
+from timescaledb.hypertables import sync as sync_module
+from timescaledb.hypertables.create import create_hypertable
+from timescaledb.hypertables.create_table import (
+    create_table_with_hypertable,
+    format_create_table_with_hypertable_sql,
+)
+from timescaledb.hypertables.schemas import HypertableCreateSchema
+from timescaledb.hypertables.sync import sync_all_hypertables
+from timescaledb.hypertables.validators import validate_chunk_time_interval
+from timescaledb.queries import time_bucket_gapfill_query
+from timescaledb.retention import sql as retention_sql
+from timescaledb.retention.add import add_retention_policy
+from timescaledb.retention.list import list_retention_policies
+from timescaledb.retention.sync import sync_retention_policies
+
+from .conftest import Metric, RetentionModel
+
+
+class QueryRecorder:
+    def __init__(self):
+        self.queries = []
+        self.commits = 0
+        self.rollbacks = 0
+
+    def execute(self, query):
+        self.queries.append(str(query))
+
+    def commit(self):
+        self.commits += 1
+
+    def rollback(self):
+        self.rollbacks += 1
+
+
+class FailingSession(QueryRecorder):
+    def execute(self, query):
+        raise SQLAlchemyError("boom")
+
+
+class _Result:
+    def __init__(self, rows):
+        self._rows = rows
+
+    def fetchall(self):
+        return self._rows
+
+
+class _Begin:
+    def __enter__(self):
+        return None
+
+    def __exit__(self, *exc):
+        return False
+
+
+class RetentionSession(QueryRecorder):
+    """Fake session that supports begin() and list/add retention flows."""
+
+    def __init__(self, existing_rows):
+        super().__init__()
+        self._existing_rows = existing_rows
+
+    def begin(self):
+        return _Begin()
+
+    def execute(self, query):
+        self.queries.append(str(query))
+        return _Result(self._existing_rows)
+
+
+class IsolatedSQLModel(sqlmodel.SQLModel):
+    metadata = MetaData()
+
+
+class PlainTimeModel(IsolatedSQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    time: datetime = Field(sa_type=sqlmodel.DateTime)
+    value: int
+
+
+class IntTimeModel(IsolatedSQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    time: int = Field(primary_key=True)
+    value: int
+
+
+class NotATableModel(sqlmodel.SQLModel):
+    foo: int
+
+
+# ---------------------------------------------------------------------------
+# defaults
+# ---------------------------------------------------------------------------
+
+
+def test_get_defaults_returns_mapping():
+    defaults = get_defaults()
+    assert defaults["TIME_COLUMN"] == "time"
+    assert "CHUNK_TIME_INTERVAL" in defaults
+
+
+# ---------------------------------------------------------------------------
+# retention/sql
+# ---------------------------------------------------------------------------
+
+
+def test_format_retention_policy_requires_drop_after():
+    with pytest.raises(ValueError, match="drop_after is required"):
+        retention_sql.format_retention_policy_sql_query("t", drop_after=None)
+
+
+def test_format_retention_policy_invalid_interval_type():
+    with pytest.raises(ValueError, match="Invalid interval type"):
+        retention_sql.format_retention_policy_sql_query("t", drop_after=1.5)
+
+
+def test_get_retention_policy_sql_query():
+    out = retention_sql.get_retention_policy_sql_query("metrics")
+    assert "policy_retention" in out
+    assert "metrics" in out
+
+
+# ---------------------------------------------------------------------------
+# retention add / list
+# ---------------------------------------------------------------------------
+
+
+def test_add_retention_policy_with_table_name():
+    session = QueryRecorder()
+    add_retention_policy(session, table_name="metrics", drop_after="1 year")
+    assert "add_retention_policy" in session.queries[0]
+
+
+def test_list_retention_policies_returns_none_when_no_rows():
+    class NoneSession:
+        def execute(self, query):
+            return self
+
+        def fetchall(self):
+            return None
+
+    assert list_retention_policies(NoneSession()) is None
+
+
+# ---------------------------------------------------------------------------
+# retention sync
+# ---------------------------------------------------------------------------
+
+
+def test_sync_retention_policies_skips_existing():
+    table = RetentionModel.__tablename__
+    session = RetentionSession(existing_rows=[(table,)])
+    sync_retention_policies(session, RetentionModel)
+    # policy already exists -> no add_retention_policy executed beyond the list query
+    assert all("add_retention_policy" not in q for q in session.queries)
+
+
+def test_sync_retention_policies_adds_when_missing():
+    session = RetentionSession(existing_rows=None)
+    sync_retention_policies(session, RetentionModel, drop_after="1 year")
+    assert any("add_retention_policy" in q for q in session.queries)
+
+
+# ---------------------------------------------------------------------------
+# hypertables/create
+# ---------------------------------------------------------------------------
+
+
+def test_create_hypertable_requires_model_or_table_name():
+    with pytest.raises(ValueError, match="model or table_name is required"):
+        create_hypertable(QueryRecorder(), model=None, table_name=None)
+
+
+def test_create_hypertable_overwrite_model_params():
+    session = QueryRecorder()
+    create_hypertable(
+        session,
+        model=Metric,
+        hypertable_options={"if_not_exists": True, "migrate_data": True},
+        overwrite_model_params=True,
+    )
+    assert session.commits == 1
+    assert "create_hypertable" in session.queries[0]
+
+
+# ---------------------------------------------------------------------------
+# hypertables/create_table
+# ---------------------------------------------------------------------------
+
+
+def test_format_create_table_requires_table_model():
+    with pytest.raises(ValueError, match="table=True"):
+        format_create_table_with_hypertable_sql(NotATableModel)
+
+
+def test_format_create_table_invalid_chunk_interval():
+    with pytest.raises(ValueError, match="Invalid chunk interval"):
+        format_create_table_with_hypertable_sql(Metric, chunk_interval=1.5)
+
+
+def test_format_create_table_minimal_options():
+    out = format_create_table_with_hypertable_sql(PlainTimeModel)
+    assert "tsdb.hypertable" in out
+    assert "tsdb.partition_column = 'time'" in out
+
+
+def test_create_table_with_hypertable_without_commit():
+    session = QueryRecorder()
+    create_table_with_hypertable(session, model=PlainTimeModel, commit=False)
+    assert session.commits == 0
+    assert len(session.queries) == 1
+
+
+# ---------------------------------------------------------------------------
+# hypertables/schemas
+# ---------------------------------------------------------------------------
+
+
+def test_hypertable_schema_invalid_interval_type_raises():
+    schema = HypertableCreateSchema(
+        table_name="t", time_column="time", chunk_time_interval=1.5
+    )
+    with pytest.raises(ValueError, match="Invalid interval type"):
+        schema.to_sql_query()
+
+
+# ---------------------------------------------------------------------------
+# hypertables/sync
+# ---------------------------------------------------------------------------
+
+
+def test_sync_all_hypertables_creates_missing(monkeypatch):
+    monkeypatch.setattr(sync_module, "list_hypertables", lambda session: [])
+    session = QueryRecorder()
+    sync_all_hypertables(session, Metric)
+    assert session.commits == 1
+    assert any("create_hypertable" in q for q in session.queries)
+
+
+def test_sync_all_hypertables_skips_existing(monkeypatch):
+    existing = type("HT", (), {"hypertable_name": Metric.__tablename__})()
+    monkeypatch.setattr(sync_module, "list_hypertables", lambda session: [existing])
+    session = QueryRecorder()
+    sync_all_hypertables(session, Metric)
+    assert all("create_hypertable" not in q for q in session.queries)
+
+
+def test_sync_all_hypertables_rolls_back_on_error(monkeypatch):
+    monkeypatch.setattr(sync_module, "list_hypertables", lambda session: [])
+    session = FailingSession()
+    with pytest.raises(SQLAlchemyError):
+        sync_all_hypertables(session, Metric)
+    assert session.rollbacks == 1
+
+
+# ---------------------------------------------------------------------------
+# hypertables/validators
+# ---------------------------------------------------------------------------
+
+
+def test_validate_chunk_time_interval_integer_column_valid():
+    validate_chunk_time_interval(IntTimeModel, "time", 1000)
+
+
+def test_validate_chunk_time_interval_integer_column_invalid():
+    with pytest.raises(InvalidChunkTimeInterval, match="must be an integer"):
+        validate_chunk_time_interval(IntTimeModel, "time", "not-an-int")
+
+
+def test_validate_chunk_time_interval_datetime_integer_invalid():
+    class BadInt(int):
+        def __int__(self):
+            raise OverflowError("too big")
+
+    with pytest.raises(InvalidChunkTimeInterval, match="must be an integer"):
+        validate_chunk_time_interval(Metric, "time", BadInt(5))
+
+
+def test_validate_chunk_time_interval_datetime_unhandled_type_is_noop():
+    # A float is neither timedelta/int/str -> falls through without raising.
+    assert validate_chunk_time_interval(Metric, "time", 1.5) is None
+
+
+# ---------------------------------------------------------------------------
+# hyperfunctions
+# ---------------------------------------------------------------------------
+
+
+def test_time_bucket_with_integer_offset():
+    expr = time_bucket("5 minutes", Metric.time, offset=5)
+    assert expr is not None
+
+
+def test_time_bucket_with_interval_offset():
+    expr = time_bucket("5 minutes", Metric.time, offset="1 hour")
+    assert expr is not None
+
+
+# ---------------------------------------------------------------------------
+# queries
+# ---------------------------------------------------------------------------
+
+
+class _FakeResult:
+    def mappings(self):
+        return self
+
+    def all(self):
+        return []
+
+
+class _CapturingSession:
+    def exec(self, query):
+        return _FakeResult()
+
+
+def test_time_bucket_gapfill_query_accepts_instrumented_attributes():
+    session = _CapturingSession()
+    result = time_bucket_gapfill_query(
+        session, Metric, time_field=Metric.time, metric_field=Metric.value
+    )
+    assert result == []