Feature/unsafe queries / Bug Fix Losing Connection

src/Connection.php

@@ -12,6 +12,7 @@
 use Illuminate\Support\Str;
 use PDPhilip\Elasticsearch\DSL\Bridge;
 use PDPhilip\Elasticsearch\DSL\Results;
+use PDPhilip\Elasticsearch\Exceptions\LogicException;
 use RuntimeException;
 
 use function array_replace_recursive;
@@ -74,10 +75,12 @@ class Connection extends BaseConnection
 
     protected mixed $elasticMetaHeader = null;
 
-    protected bool $rebuild = false;
-
     protected string $connectionName;
 
+    protected bool $byPassMapValidation = false;
+
+    protected int $insertChunkSize = 1000;
+
     /**
      * @var Query\Processor
      */
@@ -91,6 +94,7 @@ public function __construct(array $config)
         $this->config = $config;
 
         $this->_sanitizeConfig();
+
         $this->_validateConnection();
 
         $this->setOptions();
@@ -127,8 +131,34 @@ public function setOptions(): void
               : $this->config['error_log_index'];
         }
 
+        if (! empty($this->config['options']['bypass_map_validation'])) {
+            $this->byPassMapValidation = $this->config['options']['bypass_map_validation'];
+        }
+
+        if (! empty($this->config['options']['insert_chunk_size'])) {
+            $this->insertChunkSize = $this->config['options']['insert_chunk_size'];
+        }
+
+    }
+
+    /** {@inheritdoc} */
+    public function table($table, $as = null)
+    {
+        $query = new Query\Builder($this, new Query\Processor);
+
+        return $query->from($table);
+    }
+
+    /** {@inheritdoc} */
+    public function disconnect(): void
+    {
+        $this->client = null;
     }
 
+    //----------------------------------------------------------------------
+    // Getters
+    //----------------------------------------------------------------------
+
     /** {@inheritdoc} */
     public function getTablePrefix(): ?string
     {
@@ -177,21 +207,20 @@ public function getIndex(): string
         return $this->index;
     }
 
-    public function setIndex(string $index): string
+    /** {@inheritdoc} */
+    public function getDriverName(): string
     {
-        $this->index = $this->indexPrefix && ! str_contains($index, $this->indexPrefix.'_')
-          ? $this->indexPrefix.'_'.$index
-          : $index;
-
-        return $this->getIndex();
+        return 'elasticsearch';
     }
 
-    /** {@inheritdoc} */
-    public function table($table, $as = null)
+    public function getClient(): ?Client
     {
-        $query = new Query\Builder($this, new Query\Processor);
+        return $this->client;
+    }
 
-        return $query->from($table);
+    public function getMaxSize(): int
+    {
+        return $this->maxSize;
     }
 
     /**
@@ -202,73 +231,76 @@ public function getSchemaBuilder(): Schema\Builder
         return new Schema\Builder($this);
     }
 
-    /** {@inheritdoc} */
-    public function disconnect(): void
+    public function getAllowIdSort(): bool
     {
-        $this->client = null;
+        return $this->allowIdSort;
     }
 
-    /** {@inheritdoc} */
-    public function getDriverName(): string
+    public function getBypassMapValidation(): bool
     {
-        return 'elasticsearch';
+        return $this->byPassMapValidation;
     }
 
-    public function rebuildConnection(): void
+    public function getInsertChunkSize(): int
     {
-        $this->rebuild = true;
+        return $this->insertChunkSize;
     }
 
-    public function getClient(): ?Client
+    /** {@inheritdoc} */
+    protected function getDefaultPostProcessor(): Query\Processor
     {
-        return $this->client;
+        return new Query\Processor;
     }
 
-    public function getMaxSize(): int
+    /** {@inheritdoc} */
+    protected function getDefaultQueryGrammar(): Query\Grammar
     {
-        return $this->maxSize;
+        return new Query\Grammar;
     }
 
-    public function setMaxSize($value): void
+    /** {@inheritdoc} */
+    protected function getDefaultSchemaGrammar(): Schema\Grammar
     {
-        $this->maxSize = $value;
+        return new Schema\Grammar;
     }
 
-    public function getAllowIdSort(): bool
+    //----------------------------------------------------------------------
+    // Setters
+    //----------------------------------------------------------------------
+
+    public function setIndex(string $index): string
     {
-        return $this->allowIdSort;
+        $this->index = $this->indexPrefix && ! str_contains($index, $this->indexPrefix.'_')
+            ? $this->indexPrefix.'_'.$index
+            : $index;
+
+        return $this->getIndex();
+    }
+
+    public function setMaxSize($value): void
+    {
+        $this->maxSize = $value;
     }
 
     public function __call($method, $parameters)
     {
         if (! $this->index) {
             $this->index = $this->indexPrefix.'*';
         }
-        if ($this->rebuild) {
+
+        // If we are missing a database connection client we need to reconnect.
+        if (! $this->client) {
             $this->client = $this->buildConnection();
-            $this->rebuild = false;
         }
         $bridge = new Bridge($this);
 
-        return $bridge->{'process'.Str::studly($method)}(...$parameters);
-    }
+        $methodName = 'process'.Str::studly($method);
 
-    /** {@inheritdoc} */
-    protected function getDefaultPostProcessor(): Query\Processor
-    {
-        return new Query\Processor;
-    }
-
-    /** {@inheritdoc} */
-    protected function getDefaultQueryGrammar(): Query\Grammar
-    {
-        return new Query\Grammar;
-    }
+        if (! method_exists($bridge, $methodName)) {
+            throw new LogicException("{$methodName} does not exist on the bridge.");
+        }
 
-    /** {@inheritdoc} */
-    protected function getDefaultSchemaGrammar(): Schema\Grammar
-    {
-        return new Schema\Grammar;
+        return $bridge->{$methodName}(...$parameters);
     }
 
     /**
@@ -288,8 +320,10 @@ private function _sanitizeConfig(): void
                 'password' => null,
                 'api_key' => null,
                 'api_id' => null,
-                'index_prefix' => null,
+                'index_prefix' => '',
                 'options' => [
+                    'bypass_map_validation' => false, // This skips the safety checks for Elastic Specific queries.
+                    'insert_chunk_size' => 1000, // This is the maximum insert chunk size to use when bulk inserting
                     'logging' => false,
                     'allow_id_sort' => false,
                     'ssl_verification' => true,

src/DSL/Bridge.php

@@ -20,7 +20,7 @@ class Bridge
 
     protected Connection $connection;
 
-    protected Client $client;
+    protected ?Client $client;
 
     protected ?string $errorLogger;
 
@@ -1125,10 +1125,8 @@ public function parseRequiredKeywordMapping($field): ?string
             $fullMap = new Collection($mapping);
             $keywordFields = $fullMap->filter(fn ($value) => $value == 'keyword');
             $this->cachedKeywordFields = $keywordFields;
-            // Log::info('cached');
 
         }
-        // Log::info('returned');
         $keywordFields = $this->cachedKeywordFields;
 
         if ($keywordFields->isEmpty()) {
@@ -1430,7 +1428,6 @@ private function _throwError(Exception $exception, $params, $queryTag): QueryExc
         $errorMsg = $exception->getMessage();
         $errorCode = $exception->getCode();
         $queryTag = str_replace('_', '', $queryTag);
-        $this->connection->rebuildConnection();
         $error = new Results([], [], $params, $queryTag);
         $error->setError($errorMsg, $errorCode);
 

src/DSL/QueryBuilder.php

@@ -337,20 +337,28 @@ private function _parseCondition($condition, $parentField = null): array
                     $queryPart = ['bool' => ['must_not' => [['match' => [$field => $operand]]]]];
                     break;
                 case 'in':
-                    $keywordField = $this->parseRequiredKeywordMapping($field);
-                    if (! $keywordField) {
+                    if ($this->connection->getBypassMapValidation()) {
                         $queryPart = ['terms' => [$field => $operand]];
                     } else {
-                        $queryPart = ['terms' => [$keywordField => $operand]];
+                        $keywordField = $this->parseRequiredKeywordMapping($field);
+                        if (! $keywordField) {
+                            $queryPart = ['terms' => [$field => $operand]];
+                        } else {
+                            $queryPart = ['terms' => [$keywordField => $operand]];
+                        }
                     }
 
                     break;
                 case 'nin':
-                    $keywordField = $this->parseRequiredKeywordMapping($field);
-                    if (! $keywordField) {
+                    if ($this->connection->getBypassMapValidation()) {
                         $queryPart = ['bool' => ['must_not' => ['terms' => [$field => $operand]]]];
                     } else {
-                        $queryPart = ['bool' => ['must_not' => ['terms' => [$keywordField => $operand]]]];
+                        $keywordField = $this->parseRequiredKeywordMapping($field);
+                        if (! $keywordField) {
+                            $queryPart = ['bool' => ['must_not' => ['terms' => [$field => $operand]]]];
+                        } else {
+                            $queryPart = ['bool' => ['must_not' => ['terms' => [$keywordField => $operand]]]];
+                        }
                     }
 
                     break;
@@ -367,10 +375,16 @@ private function _parseCondition($condition, $parentField = null): array
                     $queryPart = ['match_phrase_prefix' => [$field => ['query' => $operand]]];
                     break;
                 case 'exact':
-                    $keywordField = $this->parseRequiredKeywordMapping($field);
-                    if (! $keywordField) {
-                        throw new ParameterException('Field ['.$field.'] is not a keyword field which is required for the [exact] operator.');
+
+                    if ($this->connection->getBypassMapValidation()) {
+                        $keywordField = $field;
+                    } else {
+                        $keywordField = $this->parseRequiredKeywordMapping($field);
+                        if (! $keywordField) {
+                            throw new ParameterException('Field ['.$field.'] is not a keyword field which is required for the [exact] operator.');
+                        }
                     }
+
                     $queryPart = ['term' => [$keywordField => $operand]];
                     break;
                 case 'group':

src/Exceptions/LaravelElasticsearchException.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PDPhilip\Elasticsearch\Exceptions;
+
+use Exception;
+
+class LaravelElasticsearchException extends Exception {}

src/Exceptions/LogicException.php

@@ -0,0 +1,7 @@
+<?php
+
+declare(strict_types=1);
+
+namespace PDPhilip\Elasticsearch\Exceptions;
+
+class LogicException extends LaravelElasticsearchException {}

src/Exceptions/MissingOrderException.php

@@ -6,7 +6,7 @@
 
 use Exception;
 
-class MissingOrderException extends Exception
+class MissingOrderException extends LaravelElasticsearchException
 {
     //    private array $_details;
 

src/Query/Builder.php

@@ -1678,7 +1678,9 @@ protected function _processInsert(array $values, bool $returnData, bool $saveWit
         }
         $this->applyBeforeQueryCallbacks();
 
-        collect($values)->chunk(1000)->each(callback: function ($chunk) use (&$response, $returnData) {
+        $insertChunkSize = $this->getConnection()->getInsertChunkSize();
+
+        collect($values)->chunk($insertChunkSize)->each(callback: function ($chunk) use (&$response, $returnData) {
             $result = $this->connection->insertBulk($chunk->toArray(), $returnData, $this->refresh);
             if ((bool) $result['hasErrors']) {
                 $response['hasErrors'] = true;

tests/Eloquent/ElasticsearchConnectionOptionsTest.php

@@ -0,0 +1,40 @@
+<?php
+
+declare(strict_types=1);
+
+use Workbench\App\Models\Post;
+use Workbench\App\Models\PostUnsafe;
+use Workbench\App\Models\Product;
+use Workbench\App\Models\ProductUnsafe;
+
+test('whereExact when Unsafe queries is used and a keyword is not specified.', function () {
+    Product::factory()->state(['name' => 'John Smith'])->create();
+
+    $products = ProductUnsafe::whereExact('name', 'John Smith')->get();
+
+    expect($products)->toBeEmpty();
+});
+
+test('whereExact when Unsafe queries is used and a keyword is specified.', function () {
+    Product::factory()->state(['name' => 'John Smith'])->create();
+
+    $products = ProductUnsafe::whereExact('name.keyword', 'John Smith')->get();
+
+    expect($products->first()->name)->toEqual('John Smith');
+});
+
+test('fails whereExact on text field.', function () {
+    Post::factory()->state(['content' => 'John Smith'])->create();
+
+    $products = Post::whereExact('content', 'John Smith')->get();
+
+    expect($products->first()->name)->toEqual('John Smith');
+})->throws(PDPhilip\Elasticsearch\DSL\exceptions\ParameterException::class);
+
+test('does not fail whereExact on text field with Unsafe queries.', function () {
+    Post::factory()->state(['content' => 'John Smith'])->create();
+
+    $posts = PostUnsafe::whereExact('content', 'John Smith')->get();
+
+    expect($posts)->toBeEmpty();
+});

tests/TestCase.php

@@ -49,5 +49,15 @@ protected function getEnvironmentSetUp($app): void
                 'logging' => true,
             ],
         ]);
+        $app['config']->set('database.connections.elasticsearch_unsafe', [
+            'driver' => 'elasticsearch',
+            'auth_type' => 'http',
+            'hosts' => ['http://localhost:9200'],
+            'options' => [
+                'bypass_map_validation' => true,
+                'insert_chunk_size' => 10000,
+                'logging' => true,
+            ],
+        ]);
     }
 }