diff --git a/lang/php/lib/Datum/AvroIOCollectionSizeException.php b/lang/php/lib/Datum/AvroIOCollectionSizeException.php
new file mode 100644
index 00000000000..6aa9b0bce0a
--- /dev/null
+++ b/lang/php/lib/Datum/AvroIOCollectionSizeException.php
@@ -0,0 +1,43 @@
+<?php
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+declare(strict_types=1);
+
+namespace Apache\Avro\Datum;
+
+use Apache\Avro\AvroException;
+
+/**
+ * Raised when an array or map declares more items than the configured maximum.
+ *
+ * The block count of an array or map is read from the (potentially untrusted or
+ * truncated) input and drives allocation of the resulting collection. This
+ * exception guards against unbounded memory allocation from a very large or
+ * malformed block count.
+ */
+class AvroIOCollectionSizeException extends AvroException
+{
+    public function __construct(int $maxItems)
+    {
+        parent::__construct(
+            sprintf('Cannot read collections larger than %d items.', $maxItems)
+        );
+    }
+}
diff --git a/lang/php/lib/Datum/AvroIODatumReader.php b/lang/php/lib/Datum/AvroIODatumReader.php
index 48f193f4c02..33b34a001d6 100644
--- a/lang/php/lib/Datum/AvroIODatumReader.php
+++ b/lang/php/lib/Datum/AvroIODatumReader.php
@@ -43,10 +43,44 @@
  */
 class AvroIODatumReader
 {
+    /**
+     * Default upper bound on the number of items in a single decoded array or
+     * map. The block count of an array or map is read from the (potentially
+     * untrusted or truncated) input and drives allocation of the resulting
+     * collection. Reading a collection that declares more items than this limit
+     * throws an {@see AvroIOCollectionSizeException} instead of attempting a
+     * potentially huge allocation. Mirrors the Java SDK's collection item limit.
+     */
+    public const DEFAULT_MAX_COLLECTION_ITEMS = 2147483639; // Integer.MAX_VALUE - 8, matching the Java SDK
+
+    /**
+     * Name of the environment variable used to override the default maximum
+     * number of items permitted in a single decoded array or map.
+     */
+    public const MAX_COLLECTION_ITEMS_ENV = 'AVRO_MAX_COLLECTION_ITEMS';
+
+    private int $maxCollectionItems;
+
     public function __construct(
         private ?AvroSchema $writersSchema = null,
         private ?AvroSchema $readersSchema = null
     ) {
+        $this->maxCollectionItems = self::defaultMaxCollectionItems();
+    }
+
+    /**
+     * Set the maximum number of items permitted in a single decoded array or map.
+     *
+     * @throws \InvalidArgumentException if $maxCollectionItems is negative
+     */
+    public function setMaxCollectionItems(int $maxCollectionItems): void
+    {
+        if ($maxCollectionItems < 0) {
+            throw new \InvalidArgumentException(
+                sprintf('maxCollectionItems must not be negative, got %d.', $maxCollectionItems)
+            );
+        }
+        $this->maxCollectionItems = $maxCollectionItems;
     }
 
     public function setWritersSchema(AvroSchema $schema): void
@@ -285,11 +319,14 @@ public function readArray(
     ): array {
         $items = [];
         $blockCount = $decoder->readLong();
-        while (0 !== $blockCount) {
+        // Loose comparison: on 32-bit builds with GMP, readLong() may return a
+        // numeric string, so a strict check against 0 would never terminate.
+        while (0 != $blockCount) {
             if ($blockCount < 0) {
                 $blockCount = -$blockCount;
                 $decoder->readLong(); // Read (and ignore) block size
             }
+            $this->checkCollectionItems(count($items), $blockCount);
             for ($i = 0; $i < $blockCount; $i++) {
                 $items[] = $this->readData(
                     $writersSchema->items(),
@@ -312,6 +349,7 @@ public function readMap(
         AvroIOBinaryDecoder $decoder
     ): array {
         $items = [];
+        $itemsRead = 0;
         $pair_count = $decoder->readLong();
         while (0 != $pair_count) {
             if ($pair_count < 0) {
@@ -320,6 +358,10 @@ public function readMap(
                 $decoder->readLong();
             }
 
+            // Track the number of decoded pairs rather than count($items): repeated
+            // keys collapse in the result and would otherwise let the cumulative
+            // check be bypassed by a stream that keeps rewriting the same key.
+            $this->checkCollectionItems($itemsRead, $pair_count);
             for ($i = 0; $i < $pair_count; $i++) {
                 $key = $decoder->readString();
                 $items[$key] = $this->readData(
@@ -328,6 +370,7 @@ public function readMap(
                     $decoder
                 );
             }
+            $itemsRead += $pair_count;
             $pair_count = $decoder->readLong();
         }
 
@@ -530,6 +573,35 @@ public function readDefaultValue(AvroSchema $fieldSchema, mixed $defaultValue):
         }
     }
 
+    private static function defaultMaxCollectionItems(): int
+    {
+        $env = getenv(self::MAX_COLLECTION_ITEMS_ENV);
+        if (false !== $env && ctype_digit($env)) {
+            return (int) $env;
+        }
+
+        return self::DEFAULT_MAX_COLLECTION_ITEMS;
+    }
+
+    /**
+     * Guard against unbounded allocation when decoding an array or map.
+     *
+     * @param int $existing  the number of items already decoded for the collection
+     * @param int $blockCount the number of items in the next block; this is the
+     *                        normalized (non-negative) count
+     *
+     * @throws AvroIOCollectionSizeException if the block count is negative or the
+     *                                       running total would exceed the maximum
+     */
+    private function checkCollectionItems(int $existing, int $blockCount): void
+    {
+        // Use subtraction rather than $existing + $blockCount so the comparison
+        // cannot overflow when the limit is configured close to PHP_INT_MAX.
+        if ($blockCount < 0 || $blockCount > $this->maxCollectionItems - $existing) {
+            throw new AvroIOCollectionSizeException($this->maxCollectionItems);
+        }
+    }
+
     private function readDecimal(string $bytes, int $scale): string
     {
         $mostSignificantBit = ord($bytes[0]) & 0x80;
diff --git a/lang/php/lib/autoload.php b/lang/php/lib/autoload.php
index de1a863513b..12c9475e794 100644
--- a/lang/php/lib/autoload.php
+++ b/lang/php/lib/autoload.php
@@ -33,6 +33,7 @@
 
 include __DIR__.'/Datum/AvroIOBinaryDecoder.php';
 include __DIR__.'/Datum/AvroIOBinaryEncoder.php';
+include __DIR__.'/Datum/AvroIOCollectionSizeException.php';
 include __DIR__.'/Datum/AvroIODatumReader.php';
 include __DIR__.'/Datum/AvroIODatumWriter.php';
 include __DIR__.'/Datum/AvroIOSchemaMatchException.php';
diff --git a/lang/php/test/IODatumReaderTest.php b/lang/php/test/IODatumReaderTest.php
index abb51507c3e..32a15edaeb7 100644
--- a/lang/php/test/IODatumReaderTest.php
+++ b/lang/php/test/IODatumReaderTest.php
@@ -22,6 +22,7 @@
 
 use Apache\Avro\Datum\AvroIOBinaryDecoder;
 use Apache\Avro\Datum\AvroIOBinaryEncoder;
+use Apache\Avro\Datum\AvroIOCollectionSizeException;
 use Apache\Avro\Datum\AvroIODatumReader;
 use Apache\Avro\Datum\AvroIODatumWriter;
 use Apache\Avro\Datum\Type\AvroDuration;
@@ -327,4 +328,104 @@ public function test_record_with_logical_types(): void
             $record
         );
     }
+
+    public function test_array_block_count_is_bounded(): void
+    {
+        $schema = AvroSchema::parse('{"type":"array","items":"null"}');
+        // A single block declaring more items than the configured limit must be
+        // rejected before allocating.
+        $decoder = $this->decoderForLongs(11, 0);
+        $reader = new AvroIODatumReader($schema, $schema);
+        $reader->setMaxCollectionItems(10);
+
+        $this->expectException(AvroIOCollectionSizeException::class);
+        $reader->read($decoder);
+    }
+
+    public function test_map_block_count_is_bounded(): void
+    {
+        $schema = AvroSchema::parse('{"type":"map","values":"null"}');
+        $decoder = $this->decoderForLongs(11, 0);
+        $reader = new AvroIODatumReader($schema, $schema);
+        $reader->setMaxCollectionItems(10);
+
+        $this->expectException(AvroIOCollectionSizeException::class);
+        $reader->read($decoder);
+    }
+
+    public function test_array_cumulative_limit_across_blocks(): void
+    {
+        $schema = AvroSchema::parse('{"type":"array","items":"null"}');
+        // Two blocks of 6 items each exceed a limit of 10 even though neither
+        // block does on its own.
+        $decoder = $this->decoderForLongs(6, 6, 0);
+        $reader = new AvroIODatumReader($schema, $schema);
+        $reader->setMaxCollectionItems(10);
+
+        $this->expectException(AvroIOCollectionSizeException::class);
+        $reader->read($decoder);
+    }
+
+    public function test_negative_block_count_is_bounded(): void
+    {
+        $schema = AvroSchema::parse('{"type":"array","items":"null"}');
+        // Negative count -11 -> 11 items, followed by a block size that is skipped.
+        $decoder = $this->decoderForLongs(-11, 0);
+        $reader = new AvroIODatumReader($schema, $schema);
+        $reader->setMaxCollectionItems(10);
+
+        $this->expectException(AvroIOCollectionSizeException::class);
+        $reader->read($decoder);
+    }
+
+    public function test_map_cumulative_limit_with_repeated_keys(): void
+    {
+        // Repeated keys collapse in the result map; the cumulative check must
+        // still count every decoded pair so the limit cannot be bypassed.
+        $schema = AvroSchema::parse('{"type":"map","values":"null"}');
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        $encoder->writeLong(6); // first block: 6 pairs, all the same key
+        for ($i = 0; $i < 6; $i++) {
+            $encoder->writeString('a'); // identical key; a null value has no bytes
+        }
+        $encoder->writeLong(6); // second block would push the total to 12 > 10
+        $decoder = new AvroIOBinaryDecoder(new AvroStringIO($io->string()));
+        $reader = new AvroIODatumReader($schema, $schema);
+        $reader->setMaxCollectionItems(10);
+
+        $this->expectException(AvroIOCollectionSizeException::class);
+        $reader->read($decoder);
+    }
+
+    public function test_array_within_limit_still_reads(): void
+    {
+        $schema = AvroSchema::parse('{"type":"array","items":"null"}');
+        $decoder = $this->decoderForLongs(3, 0); // 3 null items, then end-of-array
+        $reader = new AvroIODatumReader($schema, $schema);
+        $reader->setMaxCollectionItems(10);
+
+        $this->assertEquals([null, null, null], $reader->read($decoder));
+    }
+
+    public function test_set_max_collection_items_rejects_negative(): void
+    {
+        $reader = new AvroIODatumReader();
+        $this->expectException(\InvalidArgumentException::class);
+        $reader->setMaxCollectionItems(-1);
+    }
+
+    /**
+     * Encode the given longs (block counts / sizes / markers) into a decoder.
+     */
+    private function decoderForLongs(int ...$values): AvroIOBinaryDecoder
+    {
+        $io = new AvroStringIO();
+        $encoder = new AvroIOBinaryEncoder($io);
+        foreach ($values as $value) {
+            $encoder->writeLong($value);
+        }
+
+        return new AvroIOBinaryDecoder(new AvroStringIO($io->string()));
+    }
 }