feat: Add Catalog and Table implementations for PostgreSQL (#5487)

## Changes Made

Allows us to do
```
catalog = daft.catalog.Catalog.from_postgres("postgresql://user:pass@uri:5432/postgres")
table = catalog.get_table("myschema.mytable")
df = table.read()

...

output_table = catalog.create_table_if_not_exists("myschema.output")
output_table.write(df)
```

Depends on #5471 for
distributed writes and the `df.write_sql()` interface.

## Key decisions, limitations, and notes

- Embeddings support is currently provided via the pgvector extension
and the pgvector python client library.
- The following types are currently not supported:
  - Duration
  - Interval
- The following types currently fallback to JSONB:
  - Multidimensional lists
  - Structs
  - Maps
  - Images
  - Tensors
- No connection pooling
- Writes happen on a single node
(#5471 implements distributed
writes)
- Uses psycopg and `COPY` for writes. ADBC would potentially be a better
choice here but we ignore it for now due to its limited type support.
- Instead of assuming that `public` is the default schema, we delegate
table resolution to Postgres' [Schema Search
Path](https://www.postgresql.org/docs/current/ddl-schemas.html#DDL-SCHEMAS-PATH)

---------

Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com>

Author: desmondcheongzx

daft/catalog/__init__.py

@@ -312,6 +312,35 @@ def from_glue(
         except ImportError:
             raise ImportError("AWS Glue support not installed: pip install -U 'daft[aws]'")
 
+    @staticmethod
+    def from_postgres(connection_string: str, extensions: list[str] | None = None) -> Catalog:
+        """Create a Daft Catalog from a PostgreSQL connection string.
+
+        Note::
+            This is an experimental feature and the API may change in the future.
+
+        Args:
+            connection_string (str): a PostgreSQL connection string
+            extensions (list[str], optional): List of PostgreSQL extensions to create if they don't exist.
+                For each extension, "CREATE EXTENSION IF NOT EXISTS <extension>" will be executed.
+                Defaults to None (no extensions).
+
+        Returns:
+            Catalog: a new Catalog instance to a PostgreSQL database.
+
+        Examples:
+            >>> catalog = Catalog.from_postgres("postgresql://user:password@host:port/database")
+            >>> catalog = Catalog.from_postgres(
+            ...     "postgresql://user:password@host:port/database", extensions=["vector", "pg_stat_statements"]
+            ... )
+        """
+        try:
+            from daft.catalog.__postgres import PostgresCatalog
+
+            return PostgresCatalog.from_uri(connection_string, extensions)
+        except ImportError:
+            raise ImportError("PostgreSQL support not installed: pip install -U 'daft[postgres]'")
+
     @staticmethod
     def _from_obj(obj: object) -> Catalog:
         """Returns a Daft Catalog from a supported object type or raises a ValueError."""
@@ -326,6 +355,22 @@ def _from_obj(obj: object) -> Catalog:
             f"Unsupported catalog type: {type(obj)}; please ensure all required extra dependencies are installed."
         )
 
+    @staticmethod
+    def _validate_options(method: str, input: dict[str, Any], valid: set[str]) -> None:
+        """Validates input options against a set of valid options.
+
+        Args:
+            method (str): The method name to include in the error message
+            input (dict[str, Any]): The input options dictionary
+            valid (set[str]): Set of valid option keys
+
+        Raises:
+            ValueError: If any input options are not in the valid set
+        """
+        invalid_options = set(input.keys()) - valid
+        if invalid_options:
+            raise ValueError(f"Unsupported option(s) for {method}, found {invalid_options!s} not in {valid!s}")
+
     ###
     # create_*
     ###
@@ -796,6 +841,8 @@ def write(self, df: DataFrame, mode: Literal["append", "overwrite"] = "append",
         """
         if mode == "append":
             return self.append(df, **options)
+        else:
+            return self.overwrite(df, **options)
 
     @abstractmethod
     def append(self, df: DataFrame, **options: Any) -> None: