Skip to content

Commit daacb13

Browse files
vitkutnyvitkutny
committed
Add support for partitioning items over 400 KB limit
1 parent 474f338 commit daacb13

4 files changed

Lines changed: 239 additions & 1 deletion

File tree

‎src/Converter/PartitionItemConverter.php‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
<?php
2+
3+
namespace Rikudou\DynamoDbCache\Converter;
4+
5+
use AsyncAws\DynamoDb\DynamoDbClient;
6+
use AsyncAws\DynamoDb\Input\BatchWriteItemInput;
7+
use AsyncAws\DynamoDb\ValueObject\AttributeValue;
8+
use AsyncAws\DynamoDb\ValueObject\PutRequest;
9+
use AsyncAws\DynamoDb\ValueObject\WriteRequest;
10+
use Psr\Cache\CacheItemInterface;
11+
use Ramsey\Uuid\Uuid;
12+
use Rikudou\DynamoDbCache\Converter\CacheItemConverterInterface;
13+
use Rikudou\DynamoDbCache\DynamoCacheItem;
14+
use Rikudou\DynamoDbCache\DynamoPartitionItem;
15+
16+
final class PartitionItemConverter implements CacheItemConverterInterface
17+
{
18+
private const MAX_ITEM_SIZE_IN_BYTES = 400 * 1_024; // 400 KB
19+
20+
public function __construct(
21+
private CacheItemConverterInterface $converter,
22+
private DynamoDbClient $client,
23+
private string $tableName,
24+
private string $primaryField,
25+
private string $ttlField,
26+
private string $valueField,
27+
private bool $valueFieldBinary,
28+
) {
29+
}
30+
31+
public function supports(CacheItemInterface $cacheItem): bool
32+
{
33+
return true;
34+
}
35+
36+
public function convert(CacheItemInterface $cacheItem): DynamoCacheItem
37+
{
38+
$cacheItem = $this->converter->convert($cacheItem);
39+
$cacheItemSize = $this->calculateItemSizeInBytes($cacheItem);
40+
41+
if ($cacheItemSize <= self::MAX_ITEM_SIZE_IN_BYTES) {
42+
return $cacheItem; // partitioning not needed
43+
}
44+
45+
$maxValueSize = self::MAX_ITEM_SIZE_IN_BYTES
46+
- ($cacheItemSize - strlen($cacheItem->getRaw())) // cache item size without value
47+
- strlen(DynamoPartitionItem::formatKey('', method_exists(Uuid::class, 'uuid7') ? '00000000-0000-0000-0000-000000000000' : '00')) // partition key without prefix
48+
;
49+
50+
if ($maxValueSize <= 0) {
51+
return $cacheItem; // not enough bytes left for partitioning
52+
}
53+
54+
$partitionValues = str_split($cacheItem->getRaw(), $maxValueSize);
55+
$partitionKeys = array_map(
56+
fn (int $partitionKey) => method_exists(Uuid::class, 'uuid7') ? (string) Uuid::uuid7() : sprintf('%02d', $partitionKey),
57+
array_keys($partitionValues),
58+
);
59+
60+
$cacheItemExpiresAt = $cacheItem->getExpiresAt()?->getTimestamp();
61+
$batchWriteItemOutput = $this->client->batchWriteItem(new BatchWriteItemInput([
62+
'RequestItems' => [
63+
$this->tableName => array_map(
64+
fn (string $partitionValue, string $partitionKey) => new WriteRequest([
65+
'PutRequest' => new PutRequest([
66+
'Item' => [
67+
$this->primaryField => new AttributeValue([
68+
'S' => DynamoPartitionItem::formatKey($cacheItem->getKey(), $partitionKey),
69+
]),
70+
$this->valueField => new AttributeValue([
71+
$this->valueFieldBinary ? 'B' : 'S' => $partitionValue,
72+
]),
73+
] + ($cacheItemExpiresAt !== null ? [
74+
$this->ttlField => new AttributeValue([
75+
'N' => (string) $cacheItemExpiresAt,
76+
]),
77+
] : []),
78+
]),
79+
]),
80+
$partitionValues,
81+
$partitionKeys,
82+
),
83+
],
84+
]));
85+
86+
if ($batchWriteItemOutput->getUnprocessedItems() !== []) {
87+
return $cacheItem;
88+
}
89+
90+
return $cacheItem->set(new DynamoPartitionItem($cacheItem->getKey(), $partitionKeys));
91+
}
92+
93+
/**
94+
* @see https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/CapacityUnitCalculations.html
95+
* @see https://github.com/zaccharles/dynamodb-calculator
96+
*/
97+
private function calculateItemSizeInBytes(DynamoCacheItem $cacheItem): int
98+
{
99+
$size = strlen($this->primaryField) + strlen($cacheItem->getKey());
100+
$size += strlen($this->valueField) + strlen($cacheItem->getRaw());
101+
$expiresAt = $cacheItem->getExpiresAt();
102+
103+
if ($expiresAt !== null) {
104+
// The size of a number is approximately (number of UTF-8-encoded bytes of attribute name) + (1 byte per two significant digits) + (1 byte).
105+
$size += strlen($this->ttlField) + (int) ceil(strlen((string) $expiresAt->getTimestamp()) / 2) + 1;
106+
}
107+
108+
return $size;
109+
}
110+
}

‎src/DynamoDbCache.php‎

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,9 @@
1919
use Rikudou\Clock\ClockInterface as RikudouClock;
2020
use Rikudou\DynamoDbCache\Converter\CacheItemConverterRegistry;
2121
use Rikudou\DynamoDbCache\Converter\DefaultCacheItemConverter;
22+
use Rikudou\DynamoDbCache\Converter\PartitionItemConverter;
2223
use Rikudou\DynamoDbCache\Encoder\CacheItemEncoderInterface;
24+
use Rikudou\DynamoDbCache\Encoder\PartitionItemEncoder;
2325
use Rikudou\DynamoDbCache\Encoder\SerializeItemEncoder;
2426
use Rikudou\DynamoDbCache\Enum\NetworkErrorMode;
2527
use Rikudou\DynamoDbCache\Exception\CacheItemNotFoundException;
@@ -55,18 +57,37 @@ public function __construct(
5557
#[ExpectedValues(valuesFromClass: NetworkErrorMode::class)]
5658
private int $networkErrorMode = NetworkErrorMode::DEFAULT,
5759
private bool $valueFieldBinary = false,
60+
bool $valueFieldPartition = false,
5861
) {
5962
$clock = ClockHelper::psrClock($clock);
6063
$this->clock = $clock;
6164

6265
if ($encoder === null) {
6366
$encoder = new SerializeItemEncoder();
6467
}
68+
if ($valueFieldPartition) {
69+
$encoder = new PartitionItemEncoder(
70+
$encoder,
71+
$client,
72+
$tableName,
73+
$primaryField,
74+
$valueField,
75+
$valueFieldBinary,
76+
);
77+
}
6578
$this->encoder = $encoder;
6679

6780
if ($converter === null) {
6881
$converter = new CacheItemConverterRegistry(
69-
new DefaultCacheItemConverter($this->encoder, $this->clock)
82+
$valueFieldPartition ? new PartitionItemConverter(
83+
new DefaultCacheItemConverter($this->encoder, $this->clock),
84+
$client,
85+
$tableName,
86+
$primaryField,
87+
$ttlField,
88+
$valueField,
89+
$valueFieldBinary,
90+
) : new DefaultCacheItemConverter($this->encoder, $this->clock)
7091
);
7192
}
7293
$this->converter = $converter;

‎src/DynamoPartitionItem.php‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
<?php declare(strict_types=1);
2+
3+
namespace Rikudou\DynamoDbCache;
4+
5+
final class DynamoPartitionItem
6+
{
7+
/**
8+
* @param array<string> $keys
9+
*/
10+
public function __construct(
11+
public string $prefix,
12+
public array $keys,
13+
) {
14+
}
15+
16+
public static function formatKey(string $prefix, string $key): string
17+
{
18+
return sprintf('%s:%s', $prefix, $key);
19+
}
20+
}

‎src/Encoder/PartitionItemEncoder.php‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,87 @@
1+
<?php declare(strict_types=1);
2+
3+
namespace Rikudou\DynamoDbCache\Encoder;
4+
5+
use AsyncAws\DynamoDb\DynamoDbClient;
6+
use AsyncAws\DynamoDb\Input\BatchGetItemInput;
7+
use AsyncAws\DynamoDb\ValueObject\AttributeValue;
8+
use AsyncAws\DynamoDb\ValueObject\KeysAndAttributes;
9+
use Rikudou\DynamoDbCache\Converter\PartitionItemConverter;
10+
use Rikudou\DynamoDbCache\DynamoPartitionItem;
11+
use Rikudou\DynamoDbCache\Encoder\CacheItemEncoderInterface;
12+
use Rikudou\DynamoDbCache\Exception\CacheItemNotFoundException;
13+
14+
final readonly class PartitionItemEncoder implements CacheItemEncoderInterface {
15+
16+
17+
public function __construct(
18+
private CacheItemEncoderInterface $encoder,
19+
private DynamoDbClient $client,
20+
private string $tableName,
21+
private string $primaryField,
22+
private string $valueField,
23+
private bool $valueFieldBinary,
24+
) {
25+
}
26+
27+
/**
28+
* @see PartitionItemConverter
29+
*/
30+
public function encode(mixed $input): string
31+
{
32+
return $this->encoder->encode($input);
33+
}
34+
35+
public function decode(string $input): mixed
36+
{
37+
$value = $this->encoder->decode($input);
38+
39+
if (! $value instanceof DynamoPartitionItem) {
40+
return $value;
41+
}
42+
43+
$batchGetItemOutput = $this->client->batchGetItem(new BatchGetItemInput([
44+
'RequestItems' => [
45+
$this->tableName => new KeysAndAttributes([
46+
'Keys' => array_map(function ($key) use ($value) {
47+
return [
48+
$this->primaryField => new AttributeValue([
49+
'S' => DynamoPartitionItem::formatKey($value->prefix, $key),
50+
]),
51+
];
52+
}, $value->keys),
53+
]),
54+
],
55+
]));
56+
57+
if ($batchGetItemOutput->getUnprocessedKeys() !== []) {
58+
return throw new CacheItemNotFoundException('Unprocessed keys for partition item');
59+
}
60+
61+
$partitionedItems = $batchGetItemOutput->getResponses()[$this->tableName] ?? [];
62+
63+
if (count($partitionedItems) !== count($value->keys)) {
64+
return throw new CacheItemNotFoundException('Not all keys were found for partition item');
65+
}
66+
67+
usort(
68+
$partitionedItems,
69+
fn (array $a, array $b) => $a[$this->primaryField]->getS() <=> $b[$this->primaryField]->getS(),
70+
);
71+
72+
$originalValue = '';
73+
74+
foreach ($partitionedItems as $partitionedItem) {
75+
$partitionedItemValueField = $partitionedItem[$this->valueField] ?? null;
76+
$partitionedItemValue = $this->valueFieldBinary ? $partitionedItemValueField?->getB() : $partitionedItemValueField?->getS();
77+
78+
if ($partitionedItemValue === null) {
79+
return throw new CacheItemNotFoundException('Original value not found for partition item');
80+
}
81+
82+
$originalValue .= $partitionedItemValue;
83+
}
84+
85+
return $this->encoder->decode($originalValue);
86+
}
87+
}

0 commit comments

Comments
 (0)