Method enqueue

sdk/python/feast/feature_server.py:1004–1033

Enqueue a dataframe for offline write, grouped by push source + flags. Cheap and non-blocking; heavy I/O happens in background thread.

(
        self,
        push_source_name: str,
        df: pd.DataFrame,
        allow_registry_cache: bool,
        transform_on_write: bool,
    )

    # ---------- Public API ----------

    def enqueue(
        self,
        push_source_name: str,
        df: pd.DataFrame,
        allow_registry_cache: bool,
        transform_on_write: bool,
    ) -> None:
        """
        Enqueue a dataframe for offline write, grouped by push source + flags.
        Cheap and non-blocking; heavy I/O happens in background thread.
        """
        key = _OfflineBatchKey(
            push_source_name=push_source_name,
            allow_registry_cache=allow_registry_cache,
            transform_on_write=transform_on_write,
        )

        with self._lock:
            self._buffers[key].append(df)
            total_rows = sum(len(d) for d in self._buffers[key])
            should_flush = total_rows >= self._cfg.batch_size

        if should_flush:
            # Size-based flush
            logger.debug(
                "OfflineWriteBatcher size threshold reached for %s: %s rows",
                key,
                total_rows,
            )
            self._flush(key)

    def flush_all(self) -> None:
        """