33@runtime_checkable
34class ArrowStreamExportable(Protocol):
35    """Protocol for objects implementing the Arrow PyCapsule interface."""
36
37    def __arrow_c_stream__(self, requested_schema: Any = None) -> Any: ...

128def AlwaysFalse() -> _AlwaysFalse:  # noqa: N802
129    """Return the AlwaysFalse singleton."""
130    return _AlwaysFalse()

123def AlwaysTrue() -> _AlwaysTrue:  # noqa: N802
124    """Return the AlwaysTrue singleton."""
125    return _AlwaysTrue()

166@dataclass(frozen=True)
167class And(BooleanExpression):
168    """Logical AND with short-circuit simplification."""
169
170    left: BooleanExpression
171    right: BooleanExpression
172
173    def __new__(cls, left: BooleanExpression, right: BooleanExpression) -> BooleanExpression:  # type: ignore[misc]
174        if isinstance(left, _AlwaysTrue):
175            return right
176        if isinstance(right, _AlwaysTrue):
177            return left
178        if isinstance(left, _AlwaysFalse) or isinstance(right, _AlwaysFalse):
179            return AlwaysFalse()
180        instance = super().__new__(cls)
181        return instance
182
183    def to_sql(self) -> str:
184        return f"({self.left.to_sql()} AND {self.right.to_sql()})"
185
186    def __repr__(self) -> str:
187        return f"And(left={self.left!r}, right={self.right!r})"

61class BooleanExpression(ABC):
62    """Base for all filter expressions."""
63
64    @abstractmethod
65    def to_sql(self) -> str: ...
66
67    def __and__(self, other: BooleanExpression) -> BooleanExpression:
68        return And(self, other)
69
70    def __or__(self, other: BooleanExpression) -> BooleanExpression:
71        return Or(self, other)
72
73    def __invert__(self) -> BooleanExpression:
74        return Not(self)

199class Catalog:
200    """Ducklake catalog backed by DuckDB + ducklake extension.
201
202    Examples:
203
204    ```pycon
205    >>> import tempfile, os
206    >>> tmp = tempfile.mkdtemp()
207    >>> cat = Catalog("my_lake", os.path.join(tmp, "meta.duckdb"), data_path=os.path.join(tmp, "data"))
208    >>> cat.name
209    'my_lake'
210
211    ```
212    """
213
214    def __init__(
215        self,
216        name: str,
217        uri: str,
218        *,
219        data_path: str | None = None,
220        properties: dict[str, str] | None = None,
221        encrypted: bool = False,
222    ) -> None:
223        self._name = name
224        self._uri = uri
225        self._data_path = data_path
226        self._encrypted = encrypted
227        connection_properties, self._append_profiler = split_append_profile_properties(properties)
228
229        self._conn = duckdb.connect()
230        self._conn.execute("INSTALL ducklake; LOAD ducklake;")
231
232        # Apply connection properties (e.g., S3 configuration)
233        for key, value in connection_properties.items():
234            if not self._OPTION_KEY_RE.match(key):
235                raise ValueError(
236                    f"Invalid property key: {key!r}. Keys must contain only alphanumeric characters and underscores."
237                )
238            self._conn.execute(f"SET {key} = '{escape_string_literal(value)}'")
239
240        attach_sql = f"ATTACH 'ducklake:{uri}' AS {quote_identifier(name)}"
241        options: list[str] = []
242        if data_path is not None:
243            options.append(f"DATA_PATH '{escape_string_literal(data_path)}'")
244        if encrypted:
245            options.append("ENCRYPTED")
246        if options:
247            attach_sql += f" ({', '.join(options)})"
248        self._conn.execute(attach_sql)
249
250    @property
251    def name(self) -> str:
252        return self._name
253
254    @property
255    def encrypted(self) -> bool:
256        """Whether this catalog uses Parquet encryption."""
257        return self._encrypted
258
259    @property
260    def connection(self) -> duckdb.DuckDBPyConnection:
261        """Access the underlying DuckDB connection."""
262        return self._conn
263
264    @property
265    def append_profiler(self) -> AppendProfiler:
266        """Internal append-profiler configuration shared with catalog tables."""
267        return self._append_profiler
268
269    # -- Namespace operations ------------------------------------------------
270
271    def list_namespaces(self) -> list[str]:
272        """List all namespaces (schemas)."""
273        rows = self.fetchall(
274            "SELECT schema_name FROM information_schema.schemata "
275            f"WHERE catalog_name = '{escape_string_literal(self._name)}' "
276            "ORDER BY schema_name"
277        )
278        return [row[0] for row in rows]
279
280    def create_namespace(self, namespace: str) -> None:
281        """CREATE SCHEMA. Raises NamespaceAlreadyExistsError if exists."""
282        if self.namespace_exists(namespace):
283            raise NamespaceAlreadyExistsError(f"Namespace already exists: {namespace}")
284        self._execute(f"CREATE SCHEMA {quote_identifier(self._name)}.{quote_identifier(namespace)}")
285
286    def create_namespace_if_not_exists(self, namespace: str) -> None:
287        """Create namespace, no-op if already exists."""
288        self._execute(f"CREATE SCHEMA IF NOT EXISTS {quote_identifier(self._name)}.{quote_identifier(namespace)}")
289
290    def drop_namespace(self, namespace: str) -> None:
291        """DROP SCHEMA. Raises NamespaceNotEmptyError if non-empty, NoSuchNamespaceError if not found."""
292        if not self.namespace_exists(namespace):
293            raise NoSuchNamespaceError(f"Namespace does not exist: {namespace}")
294        tables = self.list_tables(namespace)
295        views = self.list_views(namespace)
296        if tables or views:
297            raise NamespaceNotEmptyError(f"Namespace is not empty: {namespace}")
298        self._execute(f"DROP SCHEMA {quote_identifier(self._name)}.{quote_identifier(namespace)}")
299
300    def namespace_exists(self, namespace: str) -> bool:
301        rows = self.fetchall(
302            "SELECT 1 FROM information_schema.schemata "
303            f"WHERE catalog_name = '{escape_string_literal(self._name)}' "
304            f"AND schema_name = '{escape_string_literal(namespace)}'"
305        )
306        return len(rows) > 0
307
308    # -- Table operations ----------------------------------------------------
309
310    def list_tables(self, namespace: str = "main") -> list[tuple[str, str]]:
311        """List tables in namespace. Returns list of (namespace, table_name) tuples."""
312        rows = self.fetchall(
313            "SELECT table_schema, table_name FROM information_schema.tables "
314            f"WHERE table_catalog = '{escape_string_literal(self._name)}' "
315            f"AND table_schema = '{escape_string_literal(namespace)}' "
316            "AND table_type != 'VIEW' "
317            "ORDER BY table_name"
318        )
319        return [(row[0], row[1]) for row in rows]
320
321    def create_table(
322        self,
323        identifier: str | tuple[str, str],
324        schema: Schema,
325    ) -> Table:
326        """CREATE TABLE with the given schema. Raises TableAlreadyExistsError if exists.
327
328        Examples:
329
330        ```pycon
331        >>> import tempfile, os
332        >>> from pyducklake import Catalog, Schema, required, optional, IntegerType, StringType
333        >>> tmp = tempfile.mkdtemp()
334        >>> cat = Catalog("test", os.path.join(tmp, "m.duckdb"), data_path=os.path.join(tmp, "d"))
335        >>> schema = Schema.of(required("id", IntegerType()), optional("name", StringType()))
336        >>> table = cat.create_table("users", schema)
337        >>> table.name
338        'users'
339        >>> [f.name for f in table.schema.fields]
340        ['id', 'name']
341
342        ```
343        """
344        namespace, table_name = self._resolve_identifier(identifier)
345
346        if self.table_exists((namespace, table_name)):
347            raise TableAlreadyExistsError(f"Table already exists: {namespace}.{table_name}")
348
349        fqn = self.fully_qualified_name(namespace, table_name)
350        col_defs = ", ".join(
351            f"{quote_identifier(f.name)} {ducklake_type_to_sql(f.field_type)}{' NOT NULL' if f.required else ''}"
352            for f in schema.fields
353        )
354        self._execute(f"CREATE TABLE {fqn} ({col_defs})")
355
356        # Reload schema from DuckDB to get canonical types
357        loaded_schema = self.build_schema_from_describe(namespace, table_name)
358        from pyducklake.sorting import UNSORTED
359
360        return Table(identifier=(namespace, table_name), schema=loaded_schema, catalog=self, sort_order=UNSORTED)
361
362    def create_table_if_not_exists(
363        self,
364        identifier: str | tuple[str, str],
365        schema: Schema,
366    ) -> Table:
367        """Create table if it doesn't exist, otherwise load and return existing."""
368        namespace, table_name = self._resolve_identifier(identifier)
369        if self.table_exists((namespace, table_name)):
370            return self.load_table((namespace, table_name))
371        return self.create_table((namespace, table_name), schema)
372
373    def load_table(self, identifier: str | tuple[str, str]) -> Table:
374        """Load table metadata. Raises NoSuchTableError if not found."""
375        namespace, table_name = self._resolve_identifier(identifier)
376        if not self.table_exists((namespace, table_name)):
377            raise NoSuchTableError(f"Table does not exist: {namespace}.{table_name}")
378        schema = self.build_schema_from_describe(namespace, table_name)
379        return Table(identifier=(namespace, table_name), schema=schema, catalog=self)
380
381    def drop_table(self, identifier: str | tuple[str, str]) -> None:
382        """DROP TABLE. Raises NoSuchTableError if not found."""
383        namespace, table_name = self._resolve_identifier(identifier)
384        if not self.table_exists((namespace, table_name)):
385            raise NoSuchTableError(f"Table does not exist: {namespace}.{table_name}")
386        fqn = self.fully_qualified_name(namespace, table_name)
387        self._execute(f"DROP TABLE {fqn}")
388
389    def rename_table(self, from_identifier: str | tuple[str, str], to_identifier: str | tuple[str, str]) -> Table:
390        """ALTER TABLE RENAME TO. Returns the renamed table."""
391        from_ns, from_name = self._resolve_identifier(from_identifier)
392        to_ns, to_name = self._resolve_identifier(to_identifier)
393
394        if not self.table_exists((from_ns, from_name)):
395            raise NoSuchTableError(f"Table does not exist: {from_ns}.{from_name}")
396
397        if from_ns != to_ns:
398            raise DucklakeError("Cross-namespace rename is not supported")
399
400        from_fqn = self.fully_qualified_name(from_ns, from_name)
401        # DuckDB RENAME TO expects just the new table name, not fully qualified
402        self._execute(f"ALTER TABLE {from_fqn} RENAME TO {quote_identifier(to_name)}")
403        return self.load_table((to_ns, to_name))
404
405    def table_exists(self, identifier: str | tuple[str, str]) -> bool:
406        namespace, table_name = self._resolve_identifier(identifier)
407        rows = self.fetchall(
408            "SELECT 1 FROM information_schema.tables "
409            f"WHERE table_catalog = '{escape_string_literal(self._name)}' "
410            f"AND table_schema = '{escape_string_literal(namespace)}' "
411            f"AND table_name = '{escape_string_literal(table_name)}'"
412        )
413        return len(rows) > 0
414
415    # -- View operations -----------------------------------------------------
416
417    def create_view(
418        self,
419        identifier: str | tuple[str, str],
420        sql: str,
421    ) -> View:
422        """Create a view in the catalog.
423
424        Uses: CREATE VIEW {fqn} AS {sql}
425
426        Raises ViewAlreadyExistsError if the view already exists.
427        """
428        namespace, view_name = self._resolve_identifier(identifier)
429        if self.view_exists((namespace, view_name)):
430            raise ViewAlreadyExistsError(f"View already exists: {namespace}.{view_name}")
431        fqn = self.fully_qualified_name(namespace, view_name)
432        self._execute(f"CREATE VIEW {fqn} AS {sql}")
433        return self.load_view((namespace, view_name))
434
435    def create_or_replace_view(
436        self,
437        identifier: str | tuple[str, str],
438        sql: str,
439    ) -> View:
440        """Create or replace a view. Returns the View object."""
441        namespace, view_name = self._resolve_identifier(identifier)
442        fqn = self.fully_qualified_name(namespace, view_name)
443        self._execute(f"CREATE OR REPLACE VIEW {fqn} AS {sql}")
444        return self.load_view((namespace, view_name))
445
446    def load_view(self, identifier: str | tuple[str, str]) -> View:
447        """Load a view. Raises NoSuchViewError if not found.
448
449        Queries the view's schema and SQL definition.
450        """
451        namespace, view_name = self._resolve_identifier(identifier)
452        if not self.view_exists((namespace, view_name)):
453            raise NoSuchViewError(f"View does not exist: {namespace}.{view_name}")
454        schema = self.build_schema_from_describe(namespace, view_name)
455        # Get the SQL definition
456        rows = self.fetchall(
457            "SELECT view_definition FROM information_schema.views "
458            f"WHERE table_catalog = '{escape_string_literal(self._name)}' "
459            f"AND table_schema = '{escape_string_literal(namespace)}' "
460            f"AND table_name = '{escape_string_literal(view_name)}'"
461        )
462        sql_text = rows[0][0] if rows else ""
463        return View(identifier=(namespace, view_name), schema=schema, sql=sql_text, catalog=self)
464
465    def rename_view(self, from_identifier: str | tuple[str, str], to_identifier: str | tuple[str, str]) -> View:
466        """Rename a view. Raises NoSuchViewError if not found."""
467        from_ns, from_name = self._resolve_identifier(from_identifier)
468        to_ns, to_name = self._resolve_identifier(to_identifier)
469
470        if not self.view_exists((from_ns, from_name)):
471            raise NoSuchViewError(f"View does not exist: {from_ns}.{from_name}")
472
473        if from_ns != to_ns:
474            raise DucklakeError("Cross-namespace rename is not supported")
475
476        from_fqn = self.fully_qualified_name(from_ns, from_name)
477        self._execute(f"ALTER VIEW {from_fqn} RENAME TO {quote_identifier(to_name)}")
478        return self.load_view((to_ns, to_name))
479
480    def drop_view(self, identifier: str | tuple[str, str]) -> None:
481        """Drop a view. Raises NoSuchViewError if not found."""
482        namespace, view_name = self._resolve_identifier(identifier)
483        if not self.view_exists((namespace, view_name)):
484            raise NoSuchViewError(f"View does not exist: {namespace}.{view_name}")
485        fqn = self.fully_qualified_name(namespace, view_name)
486        self._execute(f"DROP VIEW {fqn}")
487
488    def list_views(self, namespace: str = "main") -> list[tuple[str, str]]:
489        """List views in namespace. Returns list of (namespace, view_name) tuples."""
490        rows = self.fetchall(
491            "SELECT table_schema, table_name FROM information_schema.tables "
492            f"WHERE table_catalog = '{escape_string_literal(self._name)}' "
493            f"AND table_schema = '{escape_string_literal(namespace)}' "
494            "AND table_type = 'VIEW' "
495            "ORDER BY table_name"
496        )
497        return [(row[0], row[1]) for row in rows]
498
499    def view_exists(self, identifier: str | tuple[str, str]) -> bool:
500        """Check if a view exists."""
501        namespace, view_name = self._resolve_identifier(identifier)
502        rows = self.fetchall(
503            "SELECT 1 FROM information_schema.tables "
504            f"WHERE table_catalog = '{escape_string_literal(self._name)}' "
505            f"AND table_schema = '{escape_string_literal(namespace)}' "
506            f"AND table_name = '{escape_string_literal(view_name)}' "
507            "AND table_type = 'VIEW'"
508        )
509        return len(rows) > 0
510
511    # -- Internal helpers ----------------------------------------------------
512
513    def _resolve_identifier(self, identifier: str | tuple[str, str]) -> tuple[str, str]:
514        """Parse identifier into (namespace, table_name). Default namespace is 'main'."""
515        if isinstance(identifier, tuple):
516            return identifier
517        parts = identifier.split(".", 1)
518        if len(parts) == 2:
519            return (parts[0], parts[1])
520        return ("main", parts[0])
521
522    def fully_qualified_name(self, namespace: str, table_name: str) -> str:
523        """Returns 'catalog_name.namespace.table_name' with proper quoting."""
524        return f"{quote_identifier(self._name)}.{quote_identifier(namespace)}.{quote_identifier(table_name)}"
525
526    def _execute(self, sql: str, params: list[Any] | None = None) -> duckdb.DuckDBPyConnection:
527        """Execute SQL on the connection."""
528        if params:
529            return self._conn.execute(sql, params)
530        return self._conn.execute(sql)
531
532    def fetchall(self, sql: str) -> list[tuple[Any, ...]]:
533        """Execute and fetchall."""
534        result = self._conn.execute(sql)
535        rows: list[tuple[Any, ...]] = result.fetchall()
536        return rows
537
538    def build_schema_from_describe(self, namespace: str, table_name: str) -> Schema:
539        """Build a Schema from information_schema.columns for an existing table."""
540        rows = self.fetchall(
541            "SELECT column_name, data_type, is_nullable FROM information_schema.columns "
542            f"WHERE table_catalog = '{escape_string_literal(self._name)}' "
543            f"AND table_schema = '{escape_string_literal(namespace)}' "
544            f"AND table_name = '{escape_string_literal(table_name)}' "
545            "ORDER BY ordinal_position"
546        )
547        # field_id_counter tracks IDs for nested types as well
548        field_id_counter = [1]
549        fields: list[NestedField] = []
550        for col_name, col_type, is_nullable in rows:
551            dtype = _duckdb_type_to_ducklake(col_type, field_id_counter)
552            fid = field_id_counter[0]
553            field_id_counter[0] += 1
554            required = is_nullable == "NO"
555            fields.append(NestedField(field_id=fid, name=col_name, field_type=dtype, required=required))
556        return Schema(*fields)
557
558    # -- Commit Metadata -----------------------------------------------------
559
560    def set_commit_message(self, message: str, *, author: str | None = None) -> None:
561        """Set commit message and optional author for the next transaction.
562
563        Must be called inside an explicit ``BEGIN TRANSACTION`` / ``COMMIT``
564        block for the metadata to be recorded on the snapshot.
565
566        Uses: ``CALL {catalog}.set_commit_message(author, commit_message)``
567        """
568        cat = quote_identifier(self._name)
569        author_sql = f"'{escape_string_literal(author)}'" if author is not None else "NULL"
570        msg_sql = f"'{escape_string_literal(message)}'"
571        self._execute(f"CALL {cat}.set_commit_message({author_sql}, {msg_sql})")
572
573    # -- Configuration -------------------------------------------------------
574
575    _OPTION_KEY_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
576    _SCOPE_PART_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
577
578    def set_option(
579        self,
580        key: str,
581        value: str,
582        *,
583        scope: str | None = None,
584    ) -> None:
585        """Set a Ducklake configuration option.
586
587        Args:
588            key: Option key (e.g., ``'target_file_size'``).
589            value: Option value.
590            scope: Optional ``'schema.table'`` for table-level scope.
591                   ``None`` for global scope.
592
593        Uses: ``CALL {catalog}.set_option(option, value, table_name, schema)``
594
595        Raises:
596            ValueError: If ``key`` or ``scope`` contain invalid characters.
597        """
598        if not self._OPTION_KEY_RE.match(key):
599            raise ValueError(
600                f"Invalid option key: {key!r}. Keys must contain only alphanumeric characters and underscores."
601            )
602
603        cat = quote_identifier(self._name)
604        key_sql = f"'{escape_string_literal(key)}'"
605        value_sql = f"'{escape_string_literal(value)}'"
606        if scope is not None:
607            parts = scope.split(".", 1)
608            if len(parts) == 2:
609                schema_name, table_name = parts
610            else:
611                schema_name = "main"
612                table_name = parts[0]
613            for part_label, part_value in [("schema", schema_name), ("table", table_name)]:
614                if not self._SCOPE_PART_RE.match(part_value):
615                    raise ValueError(
616                        f"Invalid scope {part_label}: {part_value!r}. "
617                        "Scope parts must contain only alphanumeric characters and underscores."
618                    )
619            self._execute(
620                f"CALL {cat}.set_option("
621                f"{key_sql}, {value_sql}, "
622                f"table_name := '{escape_string_literal(table_name)}', "
623                f"schema := '{escape_string_literal(schema_name)}')"
624            )
625        else:
626            self._execute(f"CALL {cat}.set_option({key_sql}, {value_sql})")
627
628    def get_options(self) -> pa.Table:
629        """Get all configuration options as an Arrow table.
630
631        Columns: option_name, description, value, scope, scope_entry
632        """
633        cat = quote_identifier(self._name)
634        result: Any = self._conn.execute(f"SELECT * FROM {cat}.options()")
635        arrow_obj: Any = result.arrow()
636        if isinstance(arrow_obj, pa.Table):
637            return arrow_obj
638        tbl: pa.Table = arrow_obj.read_all()
639        return tbl
640
641    # -- Transaction ---------------------------------------------------------
642
643    def begin_transaction(self) -> Transaction:
644        """Begin a multi-operation transaction.
645
646        All table operations within the transaction are committed atomically.
647        """
648        from pyducklake.transaction import Transaction
649
650        return Transaction(self)
651
652    # -- Context manager -----------------------------------------------------
653
654    def __enter__(self) -> Catalog:
655        return self
656
657    def __exit__(self, *args: Any) -> None:
658        self.close()
659
660    def close(self) -> None:
661        """Close the DuckDB connection.
662
663        Best-effort DETACH of the ducklake attachment first: the ducklake
664        extension frees its per-connection catalog state on DETACH, while
665        conn.close() alone takes duckdb-core's instance-teardown path, which
666        skips that cleanup when the attached catalog carries conflict residue
667        (a transaction aborted at COMMIT) — orphaning the allocation for the
668        life of the process. A failing DETACH (already detached, dead
669        connection, transaction with outstanding work) is logged and ignored;
670        close() below is the fallback.
671        """
672        try:
673            self._conn.execute(f"DETACH {quote_identifier(self._name)}")
674        except Exception:
675            _LOG.debug("pyducklake_detach_on_close_failed", exc_info=True)
676        self._conn.close()