reading limit for query adapter

Author: ondrajodas

src/Keboola/DynamoDbExtractor/Exporter.php

@@ -69,7 +69,7 @@ public function export(): string
             $readingAdapter->read($params);
         } catch (DynamoDbException $e) {
             if ($e->getStatusCode() !== null && substr((string) $e->getStatusCode(), 0, 1) === '4') {
-                throw new UserException((string) $e->getAwsErrorCode());
+                throw new UserException((string) $e->getAwsErrorMessage());
             } else {
                 throw $e;
             }

src/Keboola/DynamoDbExtractor/ReadingAdapter/QueryReadingAdapter.php

@@ -5,6 +5,7 @@
 namespace Keboola\DynamoDbExtractor\ReadingAdapter;
 
 use Aws\DynamoDb\Marshaler;
+use Keboola\DynamoDbExtractor\ReadingLimit;
 
 class QueryReadingAdapter extends AbstractReadingAdapter
 {
@@ -19,8 +20,18 @@ public function read(array $params): void
         if (!empty($this->exportOptions['expressionAttributeNames'])) {
             $params['ExpressionAttributeNames'] = $this->exportOptions['expressionAttributeNames'];
         }
-        $response = $this->dynamoDbClient->query($params);
 
-        $this->saveResponseItems($marshaler, (array) $response['Items']);
+        $scanLimit = new ReadingLimit(1000, $this->exportOptions['limit'] ?? null);
+
+        do {
+            if (isset($response, $response['LastEvaluatedKey'])) {
+                $params['ExclusiveStartKey'] = $response['LastEvaluatedKey'];
+            }
+            $params['Limit'] = $scanLimit->getBatchSize();
+            $response = $this->dynamoDbClient->query($params)->toArray();
+            $scanLimit->decreaseLimit($response['Count']);
+
+            $this->saveResponseItems($marshaler, (array) $response['Items']);
+        } while ($scanLimit->shouldContinue() && isset($response['LastEvaluatedKey']));
     }
 }

src/Keboola/DynamoDbExtractor/ReadingAdapter/ScanReadingAdapter.php

@@ -5,7 +5,7 @@
 namespace Keboola\DynamoDbExtractor\ReadingAdapter;
 
 use Aws\DynamoDb\Marshaler;
-use Keboola\DynamoDbExtractor\ScanLimit;
+use Keboola\DynamoDbExtractor\ReadingLimit;
 
 class ScanReadingAdapter extends AbstractReadingAdapter
 {
@@ -20,7 +20,7 @@ public function read(array $params): void
             $params = array_merge($params, $paramsFromDateFilter);
         }
 
-        $scanLimit = new ScanLimit(1000, $this->exportOptions['limit'] ?? null);
+        $scanLimit = new ReadingLimit(1000, $this->exportOptions['limit'] ?? null);
 
         $marshaler = new Marshaler();
 

…/Keboola/DynamoDbExtractor/ScanLimit.php …boola/DynamoDbExtractor/ReadingLimit.phpsrc/Keboola/DynamoDbExtractor/ScanLimit.php renamed to src/Keboola/DynamoDbExtractor/ReadingLimit.php src/Keboola/DynamoDbExtractor/ScanLimit.php renamed to src/Keboola/DynamoDbExtractor/ReadingLimit.php

@@ -4,7 +4,7 @@
 
 namespace Keboola\DynamoDbExtractor;
 
-class ScanLimit
+class ReadingLimit
 {
     /** The maximum number of items to evaluate (DynamoDB's "Limit" parameter) */
     private int $batchSize;

tests/Keboola/DynamoDbExtractor/RunFullExportUnknownTableTest.php

@@ -59,7 +59,7 @@ public function testRun(): void
 
         $this->assertSame(1, $exitCode);
         $this->assertStringContainsString(
-            'ResourceNotFoundException',
+            'Cannot do operations on a non-existent table',
             $commandTester->getDisplay()
         );
     }

tests/Keboola/DynamoDbExtractor/ScanLimitTest.php

@@ -10,7 +10,7 @@ class ScanLimitTest extends TestCase
 {
     public function testLimitNull(): void
     {
-        $scanLimit = new ScanLimit(100);
+        $scanLimit = new ReadingLimit(100);
 
         $this->assertEquals(100, $scanLimit->getBatchSize());
         $this->assertTrue($scanLimit->shouldContinue());
@@ -23,7 +23,7 @@ public function testLimitNull(): void
 
     public function testLimitEqualsBatchSize(): void
     {
-        $scanLimit = new ScanLimit(100, 100);
+        $scanLimit = new ReadingLimit(100, 100);
         $scanLimit->decreaseLimit(100);
 
         $this->assertEquals(0, $scanLimit->getBatchSize());
@@ -32,7 +32,7 @@ public function testLimitEqualsBatchSize(): void
 
     public function testLimitLessThanBatchSize(): void
     {
-        $scanLimit = new ScanLimit(100, 10);
+        $scanLimit = new ReadingLimit(100, 10);
 
         $this->assertEquals(10, $scanLimit->getBatchSize());
         $this->assertTrue($scanLimit->shouldContinue());
@@ -45,7 +45,7 @@ public function testLimitLessThanBatchSize(): void
 
     public function testLimitGreaterThanBatchSize(): void
     {
-        $scanLimit = new ScanLimit(100, 223);
+        $scanLimit = new ReadingLimit(100, 223);
         $scanLimit->decreaseLimit(100);
 
         $this->assertEquals(100, $scanLimit->getBatchSize());