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)

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