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", self.config["database"])
            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.pool.acquire() as conn:
                async with 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 LIMIT and OFFSET
                            batch_query = f"""
                                SELECT *
                                FROM `{schema}`.`{table}`
                                LIMIT {BATCH_SIZE} OFFSET {offset}
                            """
                            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.pool:
                self.pool.close()
                await self.pool.wait_closed()
                self.pool = None
Generate entities for all tables in specified schemas.
backend/airweave/platform/sources/mysql.py