# imports
...

async def upsert_many(self, records: List[DrugDocumentParagraphRecord]) -> int:
    if not records:
        return 0
    total_upserted = 0
    batch_size = 50 # Thử với kích thước nhỏ trước, sau đó tăng dần khi thấy ổn định.

    for i in range(0, len(records), batch_size):
        batch = records[i:i + batch_size]
        stmt = self._build_upsert_stmt(batch)
        result = await self._execute_statement(stmt)
        total_upserted += self._get_row_count(result)

    return total_upserted

...

def _build_upsert_stmt(
        self, records: Union[NormalizedHeavyRecord, List[NormalizedHeavyRecord]]
    ) -> Executable:
    is_bulk = isinstance(records, list)
    model = NormalizedHeavyRecord

    # Trích xuất tất cả các khóa cột ngoại trừ khóa chính (`id`)
    # content_hash là khóa duy nhất cho dữ liệu, vì vậy bạn cần xây dựng và lưu nó cùng với dữ liệu
    mapper = class_mapper(model)
    column_keys = [col.key for col in mapper.columns if col.key not in ("id", "content_hash")]

    insert_stmt = pg_insert(model)
    ...
    # Loại bỏ _sa_instance_state (được thêm bởi SQLAlchemy ORM)
    ...

async def persist_doc_paragraphs():
    records = all_data
    semaphore = asyncio.Semaphore(5)
    batch_size = 300

    async def worker(batch):
        async with semaphore:
            async with factory() as session: # Cần tạo session riêng
                repository = NormalizedHeavyRecordRepository(session) 
                return await repository.upsert_many(batch)

    tasks = [worker(records[i:i + batch_size]) for i in range(0, len(records), batch_size)]
    results = await asyncio.gather(*tasks)
    return sum(results)

async def upsert_many_by_copy(self, records: List[NormalizedHeavyRecord], batch_size: int = 10000) -> int:
    ...
    # COPY ... FROM STDIN
    conn = await self.session.connection()
    raw_conn = await conn.get_raw_connection()
    pg_conn = raw_conn.driver_connection # ← Điểm cốt lõi

    # COPY vào bảng tạm
    async with pg_conn.transaction():
        tmp_table_name = "tmp_sample_data"
        target_table_name = "sample_data"
        await self.create_tmp_table(columns, pg_conn, tmp_table_name)
        await self.copy_data_to_tmp_table(batch, columns, pg_conn, tmp_table_name)
        result = await self.insert_data_to_target_table(columns, pg_conn, target_table_name, tmp_table_name)

    # Ví dụ: 'INSERT 0 5231' → trích xuất số lượng hàng
    count = int(result.split()[-1])
    return count

async def insert_data_to_target_table(self, columns, pg_conn, target_table_name, tmp_table_name):
    # Định nghĩa các cột cần ép kiểu format: column_name: TYPE(e.g. INTEGER)
    type_casts = {}
    # INSERT FROM tmp → ON CONFLICT UPSERT vào bảng chính
    select_clause = await self.build_select_clause(columns, type_casts)
    update_clause = await self.build_update_clause(columns)
    
    insert_sql = f"""
        INSERT INTO {target_table_name} ({", ".join(columns)})
        SELECT {select_clause} FROM {tmp_table_name}
        ON CONFLICT (content_hash) DO UPDATE SET {update_clause};
    """
    result = await pg_conn.execute(insert_sql)
    return result

async def copy_data_to_tmp_table(self, batch, columns, pg_conn, tmp_table_name):
    # Chuyển đổi sang CSV
    # asyncpg yêu cầu nguồn phải là đối tượng dạng bytes (không phải str)
    buffer = await self.build_csv_buffer(batch, columns) # COPY INTO bảng tạm
    await pg_conn.copy_to_table(
        table_name=tmp_table_name,
        source=buffer,
        format="text",
        columns=columns
    )

async def create_tmp_table(self, columns, pg_conn, tmp_table_name):
    await pg_conn.execute(f"""
        CREATE TEMP TABLE {tmp_table_name} (
            {', '.join(f"{col} TEXT" for col in columns)}
        ) ON COMMIT DROP;
    """)
...