apache/beam · error · ValueError

Batch size must be a positive integer

Error message

Batch size must be a positive integer

What it means

QdrantSinkConfig.__post_init__ validates that batch_size is a positive integer; zero or negative batch sizes (or non-int values that slip through typing) would produce empty or invalid upsert requests.

Solutions

1. Set batch_size to a positive integer (e.g. 64 or 128)
2. To write items one at a time, use batch_size=1 rather than 0
3. If sourced from options, coerce and validate int(value) > 0 before constructing the config

Example fix

// before
params = QdrantWriteParameters(..., batch_size=0)
// after
params = QdrantWriteParameters(..., batch_size=64)

batch_size = int(options.qdrant_batch_size)
assert batch_size > 0, f"batch_size must be > 0, got {batch_size}"

def is_valid_batch_size(n) -> bool:
    return isinstance(n, int) and not isinstance(n, bool) and n > 0

When it happens

Trigger: Constructing the sink config with batch_size=0 (often intended as 'no batching'), a negative number, or an unset option that defaults to 0 instead of the library default.

Common situations: Passing batch_size from pipeline options where the user set 0 thinking it disables batching; type confusion where a None/0 parsed from YAML becomes the value; copy-paste tuning that left batch_size=-1.

Understand the failure class

Background: "Invalid value" and "allowed values are" config errors: what your library rejected and how to fix it — this error's family across 41 libraries.

Related errors

• Collection name must be provided
• Either where_clause_fields or where_clause_value_fn must be…
• One of location, url, host, or path must be provided for…
• batch type must be pa.Table or pa.Array
• batch type must be pd.Series or pd.DataFrame

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/cc16a1f23272ea44. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/ml/rag/ingestion/qdrant.py:201

    kwargs: Additional keyword arguments to pass to the client's upsert method.
    dense_embedding_key: name for the dense vector in the qdrant collection.
    sparse_embedding_key: name for the sparse vector in the qdrant collection.
  """

  connection_params: QdrantConnectionParameters
  collection_name: str
  timeout: Optional[int] = None
  batch_size: int = DEFAULT_WRITE_BATCH_SIZE
  max_batch_byte_size: int = DEFAULT_MAX_BATCH_BYTE_SIZE
  kwargs: dict[str, Any] = field(default_factory=dict)
  dense_embedding_key: str = "dense"
  sparse_embedding_key: str = "sparse"

  def __post_init__(self):
    if not self.collection_name:
      raise ValueError("Collection name must be provided")
    if self.batch_size <= 0:
      raise ValueError("Batch size must be a positive integer")

  def create_write_transform(self) -> beam.PTransform[EmbeddableItem, Any]:
    return _QdrantWriteTransform(self)

  def create_converter(
      self,
  ) -> Callable[[EmbeddableItem], "models.PointStruct"]:
    def convert(item: EmbeddableItem) -> "models.PointStruct":
      if item.dense_embedding is None and item.sparse_embedding is None:
        raise ValueError(
            "EmbeddableItem must have at least one embedding (dense or sparse)")
      vector = {}
      if item.dense_embedding is not None:
        vector[self.dense_embedding_key] = item.dense_embedding
      if item.sparse_embedding is not None:
        sparse_indices, sparse_values = item.sparse_embedding
        vector[self.sparse_embedding_key] = models.SparseVector(
            indices=sparse_indices,

View on GitHub (pinned to 12126d8942)