Method insert_posts_batch

core/postgres_database.py:765–962

Insert multiple posts in optimized batches using PostgreSQL COPY protocol. PostgreSQL COPY is 5-10x faster than INSERT statements for bulk loading. Target: 15,000+ posts/second vs SQLite's ~3,000/second. Args: posts: List of post dictionaries with Reddit JSON st

(
        self,
        posts: list[dict[str, Any]],
        progress_callback: Callable[[int, int], None] | None = None,
        auto_tune_batch_size: bool = True,
        initial_batch_size: int = 1000,
    )

    @retry_with_exponential_backoff(max_retries=5, initial_delay=2.0, max_delay=60.0)
    def insert_posts_batch(
        self,
        posts: list[dict[str, Any]],
        progress_callback: Callable[[int, int], None] | None = None,
        auto_tune_batch_size: bool = True,
        initial_batch_size: int = 1000,
    ) -> tuple[int, int, set[str]]:
        """Insert multiple posts in optimized batches using PostgreSQL COPY protocol.

        PostgreSQL COPY is 5-10x faster than INSERT statements for bulk loading.
        Target: 15,000+ posts/second vs SQLite's ~3,000/second.

        Args:
            posts: List of post dictionaries with Reddit JSON structure
            progress_callback: Optional callback function(processed, total) for progress updates
            auto_tune_batch_size: Whether to auto-tune batch size based on performance
            initial_batch_size: Starting batch size (larger than SQLite due to better concurrency)

        Returns:
            Tuple of (successful_inserts, failed_inserts, failed_post_ids)
        """
        if not posts:
            return 0, 0, set()

        successful = 0
        failed = 0
        skipped = 0  # Track posts without valid IDs
        failed_post_ids: set[str] = set()  # Track which specific posts failed
        total_posts = len(posts)
        current_batch_size = initial_batch_size

        start_time = time.time()

        try:
            with self.pool.get_connection() as conn:
                # Disable synchronous_commit for bulk insert performance (10-20% faster)
                # Transaction safety is maintained - rollback still works correctly
                with conn.cursor() as cur:
                    cur.execute("SET LOCAL synchronous_commit = OFF")

                for batch_start in range(0, total_posts, current_batch_size):
                    batch = posts[batch_start : batch_start + current_batch_size]
                    batch_size = len(batch)

                    batch_start_time = time.time()

                    try:
                        # Use PostgreSQL COPY protocol for true streaming (no buffering)
                        # This eliminates 350MB buffer overhead per batch
                        copy_buffer = StringIO()
                        records_prepared = 0

                        for post in batch:
                            # Validate post has a valid ID before attempting insertion
                            post_id = post.get("id")
                            if not post_id or (isinstance(post_id, str) and not post_id.strip()):
                                skipped += 1
                                continue