<a id="flink-sql-text-search"></a>

# Text Search with Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® supports read-only external tables to enable full-text search
over external sources, like MongoDB, Couchbase, and Elasticsearch.

#### NOTE
Egress and ingress data transfers from Flink are offered at no cost until
3/31/2026.

Use the [TEXT_SEARCH_AGG](../../flink/reference/functions/search-functions.md#flink-sql-text-search-function) function to run text
searches over external databases in Confluent Cloud for Apache Flink.

The output of TEXT_SEARCH_AGG is an array with all rows in the external table
that have matching text in the search column.

## Syntax

```sql
SELECT * FROM text_input,
  LATERAL TABLE(TEXT_SEARCH_AGG(<external_table>, descriptor(<input_column>), <search_column>, <limit>))

-- map settings are optional
SELECT * FROM text_input,
  LATERAL TABLE(TEXT_SEARCH_AGG(<external_table>, descriptor(<input_column>), <search_column>, <limit>,
    map['async_enabled', [boolean], 'client_timeout', [int], 'max_parallelism', [int], 'retry_count', [int], 'retry_error_list', [string]]));
```

## Map options

The TEXT_SEARCH_AGG function supports several configurable options by using a
`map` parameter. Each option controls a specific aspect of the text search
operation.

Configuration
: External-table calls accept these optional parameters that control async
  behavior, parallelism, timeouts, retry policy, and debug output:
  <br/>
  - `async_enabled`: Calls to external tables are asynchronous and don’t block.
    The default is `true`.
  - `client_timeout`: Time, in seconds, after which the request to the external table
    endpoint times out. The default is 30 seconds.
  - `debug`: Return a detailed stack trace in the API response. The default
    is `false`. Confluent Cloud for Apache Flink implements data masking for error messages to
    remove any secrets or customer input, but the stack trace can contain the
    prompt itself or some part of the response string.
  - `max_parallelism`: Maximum number of parallel requests that the function
    can make. Use only when `async_enabled` is `true`. The default is 10.
  - `retry_count`: Maximum number of times Flink retries the external table
    request if the request to the external table fails. The default is 3.
  - `retry_error_list`: Comma-separated list of error codes that trigger a
    retry when the external table request fails. Only errors matching entries
    in this list cause Flink to retry the operation.

## Supported providers

The following text search providers are supported. Set the
[connector](../../flink/reference/statements/create-table.md#flink-sql-create-table-with-connector) property in the
[CREATE TABLE](../../flink/reference/statements/create-table.md#flink-sql-create-table) statement to specify the provider
for the external database table.

| Provider      | Connector value   |
|---------------|-------------------|
| Couchbase     | `couchbase`       |
| Elasticsearch | `elastic`         |
| MongoDB       | `mongodb`         |

For more information,
see [Text Search with External Databases](#flink-sql-text-search).

## Search table parameters

### Common parameters

#### {CONNECTOR}

Provider name.

#### {CONNECTOR}.PARAMS.\*

Parameters supported by external providers.

### MongoDB properties

#### mongodb.connection

MongoDB connection that includes the endpoint, username, and passwords.

This property is required.

#### mongodb.database

MongoDB database name.

This property is required.

#### mongodb.collection

MongoDB collection name.

This property is required.

#### mongodb.index

MongoDB full text search index name.

This property is required.

### Couchbase properties

#### couchbase.connection

Couchbase connection that includes the endpoint, username, and password.

This property is required.

#### couchbase.bucket

Couchbase bucket name.

This property is required.

#### couchbase.scope

Couchbase scope name.

This property is required.

#### couchbase.collection

Couchbase collection name.

This property is required.

#### couchbase.index

Couchbase search index name.

This property is required.

### Elasticsearch properties

<a id="flink-sql-ai-model-text-search-table-params-elastic-connection"></a>

#### elastic.connection

Elastic connection that includes the endpoint and authentication credentials.
For Elasticsearch Service, use an API key. For self-hosted Elasticsearch, use
username and password. For more information, see [Manage Connections](../../flink/operate-and-deploy/manage-connections.md#flink-sql-manage-connections).

This property is required.

#### elastic.index

The search index for the Elastic Search.

This property is required.

## Examples

- [Full text search with MongoDB](#flink-sql-text-search-examples-mongodb)
- [Full text search with Couchbase](#flink-sql-text-search-examples-couchbase)

<a id="flink-sql-text-search-examples-mongodb"></a>

### Full text search with MongoDB

This example shows how to run text search with MongoDB by using Flink SQL.

This example shows the following steps:

- Create a connection resource to a MongoDB database.
- Create a MongoDB external table.
- Create a text input table.
- Text search on movie plots.
- Show the results and store them in a new table.

It assumes an Atlas cluster as shown in
[Create a Cluster](https://www.mongodb.com/docs/atlas/create-database-deployment/)
and MongoDB sample dataset.

1. Run the following command to create a connection resource named
   `mongodb_connection` that uses your MongoDB credentials.
   ```sql
   CREATE CONNECTION mongodb_connection
     WITH (
       'type' = 'mongodb',
       'endpoint' = '<atlas_endpoint>',
       'username' = '<atlas_username>',
       'password' = '<atlas_password>'
     );
   ```
2. Create a MongoDB external table.
   ```sql
   -- mongodb full text search
   CREATE TABLE mongodb_movies_full_text_search (
       title STRING,
       plot STRING
   ) WITH (
       'connector' = 'mongodb',
       'mongodb.connection' = 'mongodb_connection',
       'mongodb.database' = 'sample_mflix',
       'mongodb.collection' = 'movies',
       'mongodb.index' = 'default'
   );
   ```
3. Create a search table.
   ```sql
   -- create search table
   CREATE TABLE movies_full_text_search(
     plot STRING
   );
   ```
4. Insert data into the search table.
   ```sql
   INSERT INTO movies_full_text_search VALUES
     ('A woman, with the aid of her police officer sweetheart'),
     ('A District Attorney''s outspoken stand on abortion gets him in trouble'),
     ('A tipsy doctor encounters his patient sleepwalking on a building ledge');
   ```
5. Run full text search.
   ```sql
   -- full text search
   SELECT * FROM movies_full_text_search,
     LATERAL TABLE(TEXT_SEARCH_AGG(`mongodb_movies_full_text_search`, DESCRIPTOR(plot), plot, 3));
   ```
6. Store search results in a new table.
   ```sql
   -- store search results in a new table
   CREATE TABLE movies_full_text_search_results AS
     SELECT * FROM movies_full_text_search,
       LATERAL TABLE(TEXT_SEARCH_AGG(`mongodb_movies_full_text_search`, DESCRIPTOR(plot), plot, 3));

   SELECT * FROM movies_full_text_search_results
     CROSS JOIN UNNEST(search_results)
     AS T(title, plot);
   ```

<a id="flink-sql-text-search-examples-couchbase"></a>

### Full text search with Couchbase

This example shows the following steps:

- Create a connection resource to a Couchbase database.
- Create a Couchbase external table.
- Create a text input table.
- Full text search on color descriptions.
- Select the results or stored in a new table.

Import a Couchbase example dataset by following the steps in
[Run a Vector Search with a Couchbase SDK](https://docs.couchbase.com/server/current/vector-search/run-vector-search-sdk.html).

Create a new search index to include all fields in the dataset and enable
full text search.

1. Run the following command to create a connection resource named
   “couchbase_connection” that uses your Couchbase credentials.
   ```sql
   CREATE CONNECTION couchbase_connection
     WITH (
       'type' = 'couchbase',
       'endpoint' = '<couchbase_endpoint>',
       'username' = '<couchbase_username>',
       'password' = '<couchbase_password>'
     );
   ```
2. Create a Couchbase external table.
   ```sql
   -- couchbase full text search
   CREATE TABLE couchbase_color_full_text_search (
     color STRING,
     brightness DOUBLE,
     description STRING
   ) WITH (
     'connector' = 'couchbase',
     'couchbase.connection' = 'couchbase_connection',
     'couchbase.bucket' = 'color-vector-sample',
     'couchbase.scope' = 'color',
     'couchbase.collection' = 'rgb',
     'couchbase.index' = 'rgb-vector'
   );
   ```
3. Create a text input table.
   ```sql
   CREATE TABLE text_input_couchbase_full_text_search (input STRING);
   ```
4. Insert data into the input table.
   ```sql
   INSERT INTO text_input_couchbase_full_text_search VALUES
     ('associated with growth nature and positivity'),
     ('a mix of blue and green'),
     ('the color of the sky on a clear summer day');
   ```
5. Run full text search.
   ```sql
   SELECT * FROM text_input_couchbase_full_text_search,
     LATERAL TABLE(TEXT_SEARCH_AGG(couchbase_color_full_text_search, DESCRIPTOR(description), input, 3));
   ```
6. Store search results in a new table.
   ```sql
   -- store search results in a new table
   CREATE TABLE text_input_couchbase_results AS
     SELECT * FROM text_input_couchbase_full_text_search,
       LATERAL TABLE(TEXT_SEARCH_AGG(couchbase_color_full_text_search, DESCRIPTOR(description), input, 3));

   SELECT * FROM text_input_couchbase_results
     CROSS JOIN UNNEST(search_results)
     AS T(color, brightness, description);
   ```

## Related content

- [External Tables](../../flink/concepts/external-tables.md#flink-external-tables)
- [Lookup Joins](../../flink/reference/queries/joins.md#flink-sql-lookup-joins)
- [TEXT_SEARCH_AGG function](../../flink/reference/functions/search-functions.md#flink-sql-text-search-function)
- [CREATE MODEL statement](../../flink/reference/statements/create-model.md#flink-sql-create-model)
- [AI Model Inference Functions](../../flink/reference/functions/model-inference-functions.md#flink-sql-model-inference-functions)
- [MongoDB Atlas Sink Connector](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-sink.html)
- [Elasticsearch Sink Connector](https://docs.confluent.io/cloud/current/connectors/cc-elasticsearch-service-sink.html)
- [Couchbase Sink Connector](https://docs.confluent.io/cloud/current/connectors/cc-couchbase-db-sink/cc-couchbase-db-sink.html)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).