fix(materialize tests): remove and refactor try blocks, address feedback

Author: jasonhernandez

compose.yaml

@@ -570,6 +570,8 @@ services:
       test:
         - CMD-SHELL
         - bash -c 'printf "GET / HTTP/1.1\n\n" > /dev/tcp/127.0.0.1/6875; exit $$?;'
+        - bash -c 'printf "GET / HTTP/1.1\n\n" > /dev/tcp/127.0.0.1:6876; exit $$?;'
+        - bash -c 'printf "GET / HTTP/1.1\n\n" > /dev/tcp/127.0.0.1:6874; exit $$?;'
       interval: 2s
       retries: 30
       timeout: 5s

ibis/backends/materialize/__init__.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 import sqlglot as sg
+from psycopg.errors import InternalError_, ProgrammingError, UndefinedFunction
 from sqlglot import expressions as sge
 
 import ibis
@@ -141,8 +142,13 @@ def version(self):
                         else version_str
                     )
                 return "unknown"
-        except Exception:  # noqa: BLE001
-            # Fallback to server_version if mz_version() doesn't work
+        except (
+            UndefinedFunction,
+            InternalError_,
+            ProgrammingError,
+        ):
+            # Fallback to server_version if mz_version() doesn't exist or fails
+            # Note: Materialize returns InternalError_ for undefined functions
             return super().version
 
     @property
@@ -1242,14 +1248,13 @@ def subscribe(
             while True:
                 # Fetch a batch of rows
                 cursor.execute(f"FETCH {batch_size} {cursor_name}")
-                rows = cursor.fetchall()
 
                 # Get columns from first fetch
                 if columns is None and cursor.description:
                     columns = [desc[0] for desc in cursor.description]
 
                 # If no rows and up_to was specified, we're done
-                if not rows:
+                if not (rows := cursor.fetchall()):
                     if up_to is not None:
                         break
                     # Otherwise wait a bit and try again
@@ -1939,7 +1944,7 @@ def alter_connection(
         - For SSH key rotation, update bastion server keys manually after rotation
 
         """
-        if not any([set_options, reset_options, rotate_keys]):
+        if not (set_options or reset_options or rotate_keys):
             raise ValueError(
                 "Must specify at least one of: set_options, reset_options, rotate_keys"
             )

ibis/backends/materialize/tests/conftest.py

@@ -82,27 +82,21 @@ def _load_data(self, **_: Any) -> None:
         for csv_file in self.test_files:
             table_name = csv_file.stem
             if table_name in self.connection.list_tables() and csv_file.exists():
-                try:
-                    # Get column list from schema
-                    schema = self.connection.get_schema(table_name)
-                    columns = list(schema.keys())
-                    col_list = ", ".join(f'"{c}"' for c in columns)
-
-                    # Use COPY FROM STDIN with CSV format (psycopg3 API)
-                    copy_sql = f'COPY "{table_name}" ({col_list}) FROM STDIN WITH (FORMAT CSV, HEADER true)'
-
-                    with con.cursor() as cur:
-                        # Open CSV file and use copy() context manager for psycopg3
-                        with open(csv_file) as f:
-                            with cur.copy(copy_sql) as copy:
-                                while data := f.read(8192):
-                                    copy.write(data)
-                    con.commit()
-                except Exception as e:  # noqa: BLE001
-                    # Log but don't fail - some tables might be pre-populated
-                    import warnings
-
-                    warnings.warn(f"Could not load {table_name}: {e}")
+                # Get column list from schema
+                schema = self.connection.get_schema(table_name)
+                columns = list(schema.keys())
+                col_list = ", ".join(f'"{c}"' for c in columns)
+
+                # Use COPY FROM STDIN with CSV format (psycopg3 API)
+                copy_sql = f'COPY "{table_name}" ({col_list}) FROM STDIN WITH (FORMAT CSV, HEADER true)'
+
+                with con.cursor() as cur:
+                    # Open CSV file and use copy() context manager for psycopg3
+                    with open(csv_file) as f:
+                        with cur.copy(copy_sql) as copy:
+                            while data := f.read(8192):
+                                copy.write(data)
+                con.commit()
 
     @staticmethod
     def connect(*, tmpdir, worker_id, **kw):  # noqa: ARG004