Fulltext searching in datastore query

modules/common/src/Storage/AbstractDatabaseTable.php

@@ -256,6 +256,7 @@ private function sanitizedErrorMessage(string $unsanitizedMessage) {
       // Portion of the message => User friendly message.
       'Column not found' => 'Column not found',
       'Mixing of GROUP columns' => 'You may not mix simple properties and aggregation expressions in a single query. If one of your properties includes an expression with a sum, count, avg, min or max operator, remove other properties from your query and try again',
+      'Can\'t find FULLTEXT index matching the column list' => 'You have attempted a fulltext match against a column that is not indexed for fulltext searching',
     ];
     foreach ($messages as $portion => $message) {
       if (strpos($unsanitizedMessage, $portion) !== FALSE) {

modules/common/src/Storage/SelectFactory.php

@@ -254,12 +254,40 @@ private function setQueryConditions(Query $query) {
    */
   private function addCondition($statementObj, $condition) {
     $this->normalizeOperator($condition);
+    if ($condition->operator == "match") {
+      $this->addMatchCondition($statementObj, $condition);
+      return;
+    }
     $field = (isset($condition->collection) ? $condition->collection : $this->alias)
       . '.'
       . $condition->property;
     $statementObj->condition($field, $condition->value, strtoupper($condition->operator));
   }
 
+  /**
+   * Add a custom where condition in the case of a fulltext match operator.
+   *
+   * Currently, only BOOLEAN MODE Mysql fulltext searches supported.
+   *
+   * @param \Drupal\Core\Database\Query\Select|\Drupal\Core\Database\Query\Condition $statementObj
+   *   Drupal DB API select object or condition object.
+   * @param object $condition
+   *   A condition from the DKAN query object.
+   */
+  private function addMatchCondition($statementObj, $condition) {
+    $properties = explode(',', $condition->property);
+    $fields = [];
+    foreach ($properties as $property) {
+      $fields[] = ($condition->collection ?? $this->alias)
+      . '.'
+      . $property;
+    }
+    $fields_list = implode(',', $fields);
+
+    $where = "MATCH($fields_list) AGAINST (:words IN BOOLEAN MODE)";
+    $statementObj->where($where, [':words' => $condition->value]);
+  }
+
   /**
    * Fix any quirks in DKAN query object that won't translate well to SQL.
    *

modules/common/tests/src/Unit/Storage/QueryDataProvider.php

@@ -37,6 +37,7 @@ public function getAllData($return): array {
       'likeConditionQuery',
       'containsConditionQuery',
       'startsWithConditionQuery',
+      'matchConditionQuery',
       'arrayConditionQuery',
       'nestedConditionGroupQuery',
       'sortQuery',
@@ -379,6 +380,31 @@ public static function startsWithConditionQuery($return) {
     }
   }
 
+  public static function matchConditionQuery($return) {
+    switch ($return) {
+      case self::QUERY_OBJECT:
+        $query = new Query();
+        $query->conditions = [
+          (object) [
+            "collection" => "t",
+            "property" => "field1",
+            "value" => "value",
+            "operator" => "match",
+          ],
+        ];
+        return $query;
+
+      case self::SQL:
+        return "WHERE (MATCH(t.field1) AGAINST (:words IN BOOLEAN MODE))";
+
+      case self::EXCEPTION:
+        return '';
+
+      case self::VALUES:
+        return ['value'];
+    }
+  }
+
 
   /**
    *

modules/datastore/docs/query.json

@@ -206,7 +206,8 @@
                         "in",
                         "not in",
                         "contains",
-                        "starts with"
+                        "starts with",
+                        "match"
                     ],
                     "default": "="
                  }

modules/datastore/src/Storage/DatabaseTable.php

@@ -174,12 +174,22 @@ public function innodbStrictMode(bool $on) {
   }
 
   /**
-   * Get table schema.
+   * Get the table schema in Drupal Schema API format.
    *
-   * @todo Note that this will break on PostgresSQL
+   * NOTE: This will likely fail on any db driver other than mysql.
+   *
+   * @param string $tableName
+   *   The table name.
+   * @param array $fieldsInfo
+   *   Array of fields info from DESCRIBE query.
+   *
+   * @return array
+   *   Full Drupal Schema API array.
    */
-  protected function buildTableSchema($tableName, $fieldsInfo) {
+  protected function buildTableSchema(string $tableName, array $fieldsInfo) {
+    // Add descriptions to schema from column comments.
     $canGetComment = method_exists($this->connection->schema(), 'getComment');
+    $schema = ['fields' => []];
     foreach ($fieldsInfo as $info) {
       $name = $info->Field;
       $schema['fields'][$name] = $this->translateType($info->Type, ($info->Extra ?? NULL));
@@ -188,7 +198,34 @@ protected function buildTableSchema($tableName, $fieldsInfo) {
       ];
       $schema['fields'][$name] = array_filter($schema['fields'][$name]);
     }
-    return $schema ?? ['fields' => []];
+    // Add index information to schema if available.
+    $this->addIndexInfo($schema);
+
+    return $schema;
+  }
+
+  /**
+   * Add index information to table schema.
+   *
+   * @param array $schema
+   *   Drupal Schema API array.
+   */
+  protected function addIndexInfo(array &$schema) {
+    if ($this->connection->getConnectionOptions()['driver'] != 'mysql') {
+      return;
+    }
+
+    $indexInfo = $this->connection->query("SHOW INDEXES FROM  `{$this->getTableName()}`")->fetchAll();
+    foreach ($indexInfo as $info) {
+      // Primary key is handled elsewhere.
+      if ($info->Key_name == 'PRIMARY') {
+        continue;
+      }
+      // Deviating slightly from Drupal Schema API to specify fulltext indexes.
+      $indexes_key = $info->Index_type == 'FULLTEXT' ? 'fulltext indexes' : 'indexes';
+      $name = $info->Key_name;
+      $schema[$indexes_key][$name][] = $info->Column_name;
+    }
   }
 
   /**

modules/datastore/tests/src/Unit/Storage/DatabaseTableTest.php

@@ -39,8 +39,10 @@ public function testConstruction() {
    *
    */
   public function testGetSchema() {
+    $connectionChain = $this->getConnectionChain();
+
     $databaseTable = new DatabaseTable(
-      $this->getConnectionChain()->getMock(),
+      $connectionChain->getMock(),
       $this->getResource()
     );
 
@@ -67,6 +69,17 @@ public function testGetSchema() {
           "mysql_type" => "text",
         ],
       ],
+      "indexes" => [
+        "idx1" => [
+          "first_name",
+        ],
+      ],
+      "fulltext indexes" => [
+        "ftx1" => [
+          "first_name",
+          "last_name",
+        ],
+      ],
     ];
 
     $this->assertEquals($expectedSchema['fields'], $schema['fields']);
@@ -414,6 +427,27 @@ public function testQueryColumnNotFound() {
     $databaseTable->query($query);
   }
 
+  /**
+   *
+   */
+  public function testNoFulltextIndexFound() {
+    $query = new Query();
+
+    $connectionChain = $this->getConnectionChain()
+      ->add(Connection::class, 'select', Select::class, 'select_1')
+      ->add(Select::class, 'fields', Select::class)
+      ->add(Select::class, 'condition', Select::class)
+      ->add(Select::class, 'execute', new DatabaseExceptionWrapper("SQLSTATE[HY000]: General error: 1191 Can't find FULLTEXT index matching the column list..."));
+
+    $databaseTable = new DatabaseTable(
+      $connectionChain->getMock(),
+      $this->getResource()
+    );
+
+    $this->expectExceptionMessage("You have attempted a fulltext match against a column that is not indexed for fulltext searching");
+    $databaseTable->query($query);
+  }
+
   /**
    * Private.
    */
@@ -435,11 +469,32 @@ private function getConnectionChain() {
       ]
     ];
 
+    $indexInfo = [
+      (object) [
+        'Key_name' => "idx1",
+        'Column_name' => 'first_name',
+        'Index_type' => 'FOO',
+      ],
+      (object) [
+        'Key_name' => "ftx1",
+        'Column_name' => 'first_name',
+        'Index_type' => 'FULLTEXT',
+      ],
+      (object) [
+        'Key_name' => "ftx2",
+        'Column_name' => 'first_name',
+        'Index_type' => 'FULLTEXT',
+      ],
+    ];
+
     $chain = (new Chain($this))
       // Construction.
       ->add(Connection::class, "schema", Schema::class)
       ->add(Connection::class, 'query', StatementWrapper::class)
-      ->add(StatementWrapper::class, 'fetchAll', $fieldInfo)
+      ->add(Connection::class, 'getConnectionOptions', ['driver' => 'mysql'])
+      ->add(StatementWrapper::class, 'fetchAll',
+        (new Sequence())->add($fieldInfo)->add($indexInfo)
+      )
       ->add(Schema::class, "tableExists", TRUE)
       ->add(Schema::class, 'getComment',
         (new Sequence())->add(NULL)->add('First Name')->add('lAST nAME')