async def generate_entities(self) -> AsyncGenerator[ChunkEntity, None]:
        """Generate entities for all tables in specified schemas."""
        try:
            await self._connect()

            schema = self.config.get("schema", "dbo")
            tables_config = self.config.get("tables", "*")

            # Handle both wildcard and CSV list of tables
            if tables_config == "*":
                tables = await self._get_tables(schema)
            else:
                # Split by comma and strip whitespace
                tables = [t.strip() for t in tables_config.split(",")]
                # Validate that all specified tables exist
                available_tables = await self._get_tables(schema)
                invalid_tables = set(tables) - set(available_tables)
                if invalid_tables:
                    raise ValueError(
                        f"Tables not found in schema '{schema}': {', '.join(invalid_tables)}"
                    )

            async with self.conn.cursor() as cursor:
                for table in tables:
                    # Create entity class if not already created
                    if f"{schema}.{table}" not in self.entity_classes:
                        self.entity_classes[f"{schema}.{table}"] = await self._create_entity_class(
                            schema, table
                        )

                    entity_class = self.entity_classes[f"{schema}.{table}"]

                    # Fetch and yield data
                    BATCH_SIZE = 50
                    offset = 0

                    while True:
                        # Fetch records in batches using OFFSET FETCH
                        batch_query = f"""
                            SELECT *
                            FROM [{schema}].[{table}]
                            ORDER BY (SELECT NULL)
                            OFFSET {offset} ROWS
                            FETCH NEXT {BATCH_SIZE} ROWS ONLY
                        """
                        await cursor.execute(batch_query)
                        records = await cursor.fetchall()

                        # Break if no more records
                        if not records:
                            break

                        # Get column names from cursor description
                        columns = [column[0] for column in cursor.description]

                        # Process the batch
                        for record in records:
                            # Convert record to dictionary using column names
                            data = dict(zip(columns, record, strict=False))
                            model_fields = entity_class.model_fields
                            primary_keys = model_fields["primary_key_columns"].default_factory()
                            pk_values = [str(data[pk]) for pk in primary_keys]
                            entity_id = f"{schema}.{table}:" + ":".join(pk_values)

                            entity = entity_class(entity_id=entity_id, **data)
                            yield entity

                        # Increment offset for next batch
                        offset += BATCH_SIZE

        finally:
            if self.conn:
                await self.conn.close()
                self.conn = None
Generate entities for all tables in specified schemas.
backend/airweave/platform/sources/sql_server.py