Merge pull request #24 from keboola/CM-792-ondra

Fix Incremental Fetching

.github/workflows/push.yml

@@ -22,7 +22,7 @@ jobs:
         steps:
             -
                 name: 'Check out the repo'
-                uses: actions/checkout@v2
+                uses: actions/checkout@v3
             -
                 name: 'Print Docker version'
                 run: 'docker -v'
@@ -40,8 +40,8 @@ jobs:
                     TAG="${GITHUB_REF##*/}"
                     IS_SEMANTIC_TAG=$(echo "$TAG" | grep -q '^v\?[0-9]\+\.[0-9]\+\.[0-9]\+$' && echo true || echo false)
                     echo "Tag = '$TAG', is semantic tag = '$IS_SEMANTIC_TAG'"
-                    echo "::set-output name=app_image_tag::$TAG"
-                    echo "::set-output name=is_semantic_tag::$IS_SEMANTIC_TAG"
+                    echo "is_semantic_tag=$IS_SEMANTIC_TAG" >> $GITHUB_OUTPUT
+                    echo "app_image_tag=$TAG" >> $GITHUB_OUTPUT
             -
                 name: 'Push image to ECR'
                 uses: keboola/action-push-to-ecr@master
@@ -63,7 +63,7 @@ jobs:
         steps:
             -
                 name: 'Check out the repo'
-                uses: actions/checkout@v2
+                uses: actions/checkout@v3
             -
                 name: 'Pull image from ECR'
                 uses: keboola/action-pull-from-ecr@master

src/Export.php

@@ -183,22 +183,34 @@ public static function buildIncrementalFetchingParams(
     public function getLastFetchedValue(): mixed
     {
         // Limit can be disabled with empty string
+        $lastValueOptions = [
+            $this->exportOptions->getLastValueOptions(),
+            [
+                'limit' => 1,
+                'sort' => json_encode([$this->exportOptions->getIncrementalFetchingColumn() => -1]),
+            ],
+        ];
+        foreach ($lastValueOptions as $lastValueOption) {
+            $options = array_merge(
+                $this->connectionOptions,
+                $this->exportOptions->toArray(),
+                $lastValueOption,
+            );
+
+            $cliCommand = $this->exportCommandFactory->create($options);
+            $process = Process::fromShellCommandline($cliCommand, null, null, null, null);
+            try {
+                $process->mustRun();
+            } catch (ProcessFailedException $e) {
+                $this->handleMongoExportFails($e);
+            }
 
-        $options = array_merge(
-            $this->connectionOptions,
-            $this->exportOptions->toArray(),
-            $this->exportOptions->getLastValueOptions()
-        );
-
-        $cliCommand = $this->exportCommandFactory->create($options);
-        $process = Process::fromShellCommandline($cliCommand, null, null, null, null);
-        try {
-            $process->mustRun();
-        } catch (ProcessFailedException $e) {
-            $this->handleMongoExportFails($e);
+            $output = $process->getOutput();
+            if (!empty($output)) {
+                break;
+            }
         }
 
-        $output = $process->getOutput();
         if (!empty($output)) {
             // Replace e.g. {"$date":"DATE"} to "ISODate("DATE")"
             $output = ExportHelper::convertSpecialColumnsToString($output);

tests/functional/incremental-fetching-large-limit-state/expected-code

@@ -0,0 +1 @@
+0

tests/functional/incremental-fetching-large-limit-state/expected-stdout

@@ -0,0 +1,3 @@
+Connected to mongodb://mongodb:27017/test
+Exporting "incremental"
+Done "incremental", parsed 4 records in total

tests/functional/incremental-fetching-large-limit-state/expected/data/out/state.json

@@ -0,0 +1 @@
+{"lastFetchedRow": 4}

tests/functional/incremental-fetching-large-limit-state/expected/data/out/tables/incremental.csv

@@ -0,0 +1,5 @@
+"id","decimal","date","timestamp"
+"1","123.344","2020-05-18T16:00:00Z","1587646020"
+"2","133.444","2020-02-15T13:00:00Z","1587626020"
+"3","783.028","2020-05-18T11:00:00Z","1587606020"
+"4","283.473","2020-04-18T16:00:00Z","1587146020"

tests/functional/incremental-fetching-large-limit-state/expected/data/out/tables/incremental.csv.manifest

@@ -0,0 +1 @@
+{"primary_key":null,"incremental":true}

tests/functional/incremental-fetching-large-limit-state/setUp.php

@@ -0,0 +1,12 @@
+<?php
+
+declare(strict_types = 1);
+
+use MongoExtractor\FunctionalTests\DatadirTest;
+use MongoExtractor\Tests\Traits\ImportDatasetTrait;
+
+return static function (DatadirTest $test): void {
+    (new class {
+        use ImportDatasetTrait;
+    })::importDatatasetNoAuthDb('incremental', 'dataset-incremental-fetching.json');
+};

tests/functional/incremental-fetching-large-limit-state/source/data/config.json

@@ -0,0 +1,20 @@
+{
+  "parameters": {
+    "db": {
+      "host": "mongodb",
+      "port": 27017,
+      "database": "test"
+    },
+    "tableName": "incremental",
+    "collection": "incremental",
+    "incremental": true,
+    "incrementalFetchingColumn": "id",
+    "limit": "1000",
+    "mapping": {
+      "id": "id",
+      "decimal": "decimal",
+      "date.$date": "date",
+      "timestamp": "timestamp"
+    }
+  }
+}