Merge pull request #27 from getindata/adapt-source-to-elastic

Adapt Http Lookup Source to Elastic

Author: swtwsk

CHANGELOG.md

@@ -13,6 +13,15 @@ New properties are:
   - `gid.connector.http.security.cert.client` - path to connector's certificate.
   - `gid.connector.http.security.key.client` - path to connector's private key.
   - `gid.connector.http.security.cert.server.allowSelfSigned` - allowing for self-signed certificates without adding them to KeyStore (not recommended for a production).
+- Add [LookupQueryCreator](src/main/java/com/getindata/connectors/http/LookupQueryCreator.java) and
+  [LookupQueryCreatorFactory](src/main/java/com/getindata/connectors/http/LookupQueryCreatorFactory.java) interfaces 
+  (along with a "default"
+  [GenericGetQueryCreator](src/main/java/com/getindata/connectors/http/internal/table/lookup/querycreators/GenericGetQueryCreator.java)
+  implementation) for customization of queries prepared by Lookup Source for its HTTP requests.
+- Add [ElasticSearchLiteQueryCreator](src/main/java/com/getindata/connectors/http/internal/table/lookup/querycreators/ElasticSearchLiteQueryCreator.java)
+  that prepares [`q` parameter query](https://www.elastic.co/guide/en/elasticsearch/reference/current/search-search.html#search-api-query-params-q)
+  using Lucene query string syntax (in first versions of ElasticSearch called
+  [Search _Lite_](https://www.elastic.co/guide/en/elasticsearch/guide/current/search-lite.html)).
 
 ## [0.4.0] - 2022-08-31
 

README.md

@@ -113,6 +113,18 @@ CREATE TABLE http-lookup (
 )
 ```
 
+#### Custom REST query
+Http Lookup Source builds queries out of `JOIN` clauses. One can customize how those queries are built by implementing
+[LookupQueryCreator](src/main/java/com/getindata/connectors/http/LookupQueryCreator.java) and
+[LookupQueryCreatorFactory](src/main/java/com/getindata/connectors/http/LookupQueryCreatorFactory.java) interfaces.
+Custom implementations of `LookupQueryCreatorFactory` can be registered along other factories in
+`resources/META-INF.services/org.apache.flink.table.factories.Factory` file and then referenced by their identifiers in
+the Http Lookup Source DDL property field `gid.connector.http.source.lookup.query-creator`.
+
+A default implementation that builds an "ordinary" GET query, i.e. adds `?joinColumn1=value1&joinColumn2=value2&...`
+to the URI of the endpoint,
+([GenericGetQueryCreator](src/main/java/com/getindata/connectors/http/internal/table/lookup/querycreators/GenericGetQueryCreator.java))
+is provided and set as a default.
 
 ### HTTP Sink
 The following example shows the minimum Table API example to create a [HttpDynamicSink](src/main/java/com/getindata/connectors/http/internal/table/HttpDynamicSink.java) that writes JSON values to an HTTP endpoint using POST method, assuming Flink has JAR of [JSON serializer](https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/formats/json/) installed:

src/main/java/com/getindata/connectors/http/internal/table/lookup/LookupArg.java renamed to src/main/java/com/getindata/connectors/http/LookupArg.java

@@ -1,4 +1,4 @@
-package com.getindata.connectors.http.internal.table.lookup;
+package com.getindata.connectors.http;
 
 import lombok.Data;
 import lombok.RequiredArgsConstructor;

src/main/java/com/getindata/connectors/http/LookupQueryCreator.java

@@ -0,0 +1,22 @@
+package com.getindata.connectors.http;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * An interface for a creator of a lookup query in the Http Lookup Source (e.g., the query that
+ * gets appended to the URI in GET request).
+ *
+ * <p>One can customize how those queries are built by implementing {@link LookupQueryCreator} and
+ * {@link LookupQueryCreatorFactory}.
+ */
+public interface LookupQueryCreator extends Serializable {
+    /**
+     * Create a lookup query (like the query appended to path in GET request)
+     * out of the provided arguments.
+     *
+     * @param params the list of {@link LookupArg} containing request parameters.
+     * @return a lookup query.
+     */
+    String createLookupQuery(List<LookupArg> params);
+}

src/main/java/com/getindata/connectors/http/LookupQueryCreatorFactory.java

@@ -0,0 +1,37 @@
+package com.getindata.connectors.http;
+
+import org.apache.flink.table.factories.Factory;
+
+import com.getindata.connectors.http.internal.table.lookup.HttpLookupTableSource;
+
+/**
+ * The {@link Factory} that dynamically creates and injects {@link LookupQueryCreator} to
+ * {@link HttpLookupTableSource}.
+ *
+ * <p>Custom implementations of {@link LookupQueryCreatorFactory} can be registered along other
+ * factories in <pre>resources/META-INF.services/org.apache.flink.table.factories.Factory</pre>
+ * file and then referenced by their identifiers in the HttpLookupSource DDL property field
+ * <i>gid.connector.http.source.lookup.query-creator</i>.
+ *
+ * <p>The following example shows the minimum Table API example to create a
+ * {@link HttpLookupTableSource} that uses a custom query creator created by a factory that returns
+ * <i>my-query-creator</i> as its identifier.
+ *
+ * <pre>{@code
+ * CREATE TABLE http (
+ *   id bigint,
+ *   some_field string
+ * ) WITH (
+ *   'connector' = 'rest-lookup',
+ *   'format' = 'json',
+ *   'url' = 'http://example.com/myendpoint',
+ *   'gid.connector.http.source.lookup.query-creator' = 'my-query-creator'
+ * )
+ * }</pre>
+ */
+public interface LookupQueryCreatorFactory extends Factory {
+    /**
+     * @return {@link LookupQueryCreator} custom lookup query creator instance
+     */
+    LookupQueryCreator createLookupQueryCreator();
+}

src/main/java/com/getindata/connectors/http/internal/PollingClient.java

@@ -3,7 +3,7 @@
 import java.util.List;
 import java.util.Optional;
 
-import com.getindata.connectors.http.internal.table.lookup.LookupArg;
+import com.getindata.connectors.http.LookupArg;
 
 /**
  * A client that is used to get enrichment data from external component.

src/main/java/com/getindata/connectors/http/internal/PollingClientFactory.java

@@ -4,12 +4,14 @@
 
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 
+import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.table.lookup.HttpLookupConfig;
 
 public interface PollingClientFactory<OUT> extends Serializable {
 
     PollingClient<OUT> createPollClient(
         HttpLookupConfig options,
-        DeserializationSchema<OUT> schemaDecoder
+        DeserializationSchema<OUT> schemaDecoder,
+        LookupQueryCreator lookupQueryCreator
     );
 }

src/main/java/com/getindata/connectors/http/internal/config/HttpConnectorConfigConstants.java

@@ -42,6 +42,9 @@ public final class HttpConnectorConfigConstants {
     public static final String SINK_REQUEST_CALLBACK_IDENTIFIER =
         GID_CONNECTOR_HTTP + "sink.request-callback";
 
+    public static final String SOURCE_LOOKUP_QUERY_CREATOR_IDENTIFIER =
+        GID_CONNECTOR_HTTP + "source.lookup.query-creator";
+
     // -------------- HTTPS security settings --------------
     public static final String ALLOW_SELF_SIGNED =
         GID_CONNECTOR_HTTP + "security.cert.server.allowSelfSigned";

src/main/java/com/getindata/connectors/http/internal/table/lookup/HttpLookupConnectorOptions.java

@@ -3,6 +3,9 @@
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 
+import com.getindata.connectors.http.internal.table.lookup.querycreators.GenericGetQueryCreatorFactory;
+import static com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants.SOURCE_LOOKUP_QUERY_CREATOR_IDENTIFIER;
+
 public class HttpLookupConnectorOptions {
 
     public static final ConfigOption<String> URL =
@@ -22,4 +25,9 @@ public class HttpLookupConnectorOptions {
             .booleanType()
             .defaultValue(false)
             .withDescription("Whether to use Sync and Async polling mechanism");
+
+    public static final ConfigOption<String> LOOKUP_QUERY_CREATOR_IDENTIFIER =
+        ConfigOptions.key(SOURCE_LOOKUP_QUERY_CREATOR_IDENTIFIER)
+            .stringType()
+            .defaultValue(GenericGetQueryCreatorFactory.IDENTIFIER);
 }

src/main/java/com/getindata/connectors/http/internal/table/lookup/HttpLookupTableSource.java

@@ -15,6 +15,7 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 
+import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.PollingClientFactory;
 import com.getindata.connectors.http.internal.table.lookup.HttpTableLookupFunction.ColumnData;
 
@@ -31,6 +32,8 @@ public class HttpLookupTableSource
 
     private final DecodingFormat<DeserializationSchema<RowData>> decodingFormat;
 
+    private final LookupQueryCreator lookupQueryCreator;
+
     @Override
     public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
         String[] keyNames = new String[context.getKeys().length];
@@ -50,7 +53,8 @@ public DynamicTableSource copy() {
             physicalRowDataType,
             pollingClientFactory,
             lookupConfig,
-            decodingFormat
+            decodingFormat,
+            lookupQueryCreator
         );
     }
 
@@ -80,6 +84,7 @@ private LookupRuntimeProvider buildLookupFunction(String[] keyNames, LookupConte
                 .pollingClientFactory(pollingClientFactory)
                 .schemaDecoder(schemaDecoder)
                 .columnData(columnData)
+                .lookupQueryCreator(lookupQueryCreator)
                 .options(lookupConfig)
                 .build();
 

src/main/java/com/getindata/connectors/http/internal/table/lookup/HttpLookupTableSourceFactory.java

@@ -25,12 +25,11 @@
 import static org.apache.flink.table.api.DataTypes.FIELD;
 import static org.apache.flink.table.types.utils.DataTypeUtils.removeTimeAttribute;
 
+import com.getindata.connectors.http.LookupQueryCreatorFactory;
 import com.getindata.connectors.http.internal.PollingClientFactory;
 import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
 import com.getindata.connectors.http.internal.utils.ConfigUtils;
-import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.ASYNC_POLLING;
-import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.URL;
-import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.URL_ARGS;
+import static com.getindata.connectors.http.internal.table.lookup.HttpLookupConnectorOptions.*;
 
 public class HttpLookupTableSourceFactory implements DynamicTableSourceFactory {
 
@@ -61,6 +60,13 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 FactoryUtil.FORMAT
             );
 
+        final LookupQueryCreatorFactory lookupQueryCreatorFactory =
+            FactoryUtil.discoverFactory(
+                context.getClassLoader(),
+                LookupQueryCreatorFactory.class,
+                readableConfig.get(LOOKUP_QUERY_CREATOR_IDENTIFIER)
+            );
+
         PollingClientFactory<RowData> pollingClientFactory = new JavaNetHttpPollingClientFactory();
         HttpLookupConfig lookupConfig = getHttpLookupOptions(context, readableConfig);
 
@@ -73,7 +79,8 @@ public DynamicTableSource createDynamicTableSource(Context context) {
             physicalRowDataType,
             pollingClientFactory,
             lookupConfig,
-            decodingFormat
+            decodingFormat,
+            lookupQueryCreatorFactory.createLookupQueryCreator()
         );
     }
 

src/main/java/com/getindata/connectors/http/internal/table/lookup/HttpTableLookupFunction.java

@@ -19,6 +19,8 @@
 import org.apache.flink.table.functions.FunctionContext;
 import org.apache.flink.table.functions.TableFunction;
 
+import com.getindata.connectors.http.LookupArg;
+import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.PollingClient;
 import com.getindata.connectors.http.internal.PollingClientFactory;
 
@@ -29,6 +31,8 @@ public class HttpTableLookupFunction extends TableFunction<RowData> {
 
     private final DeserializationSchema<RowData> schemaDecoder;
 
+    private final LookupQueryCreator lookupQueryCreator;
+
     @Getter
     private final ColumnData columnData;
 
@@ -44,19 +48,22 @@ private HttpTableLookupFunction(
         PollingClientFactory<RowData> pollingClientFactory,
         DeserializationSchema<RowData> schemaDecoder,
         ColumnData columnData,
-        HttpLookupConfig options) {
+        HttpLookupConfig options,
+        LookupQueryCreator lookupQueryCreator) {
 
         this.pollingClientFactory = pollingClientFactory;
         this.schemaDecoder = schemaDecoder;
         this.columnData = columnData;
         this.options = options;
+        this.lookupQueryCreator = lookupQueryCreator;
     }
 
     @Override
     public void open(FunctionContext context) throws Exception {
         super.open(context);
         this.localHttpCallCounter = new AtomicInteger(0);
-        this.client = pollingClientFactory.createPollClient(options, schemaDecoder);
+        this.client = pollingClientFactory
+            .createPollClient(options, schemaDecoder, lookupQueryCreator);
 
         context
             .getMetricGroup()

src/main/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClient.java

@@ -19,6 +19,8 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.util.StringUtils;
 
+import com.getindata.connectors.http.LookupArg;
+import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.PollingClient;
 import com.getindata.connectors.http.internal.config.HttpConnectorConfigConstants;
@@ -43,17 +45,21 @@ public class JavaNetHttpPollingClient implements PollingClient<RowData> {
 
     private final HttpLookupConfig options;
 
+    private final LookupQueryCreator lookupQueryCreator;
+
     private final String[] headersAndValues;
 
     public JavaNetHttpPollingClient(
             HttpClient httpClient,
             DeserializationSchema<RowData> runtimeDecoder,
             HttpLookupConfig options,
+            LookupQueryCreator lookupQueryCreator,
             HeaderPreprocessor headerPreprocessor) {
 
         this.httpClient = httpClient;
         this.runtimeDecoder = runtimeDecoder;
         this.options = options;
+        this.lookupQueryCreator = lookupQueryCreator;
 
         var headerMap = HttpHeaderUtils
             .prepareHeaderMap(
@@ -96,7 +102,9 @@ private Optional<RowData> queryAndProcess(List<LookupArg> params) throws Excepti
     }
 
     private HttpRequest buildHttpRequest(List<LookupArg> params) throws URISyntaxException {
-        URI uri = buildUri(params);
+        var lookupQuery = lookupQueryCreator.createLookupQuery(params);
+        URI uri = new URIBuilder(options.getUrl() + "?" + lookupQuery).build();
+
         Builder requestBuilder = HttpRequest.newBuilder()
             .uri(uri).GET()
             .timeout(Duration.ofMinutes(2));
@@ -108,16 +116,6 @@ private HttpRequest buildHttpRequest(List<LookupArg> params) throws URISyntaxExc
         return requestBuilder.build();
     }
 
-    private URI buildUri(List<LookupArg> params) throws URISyntaxException {
-
-        URIBuilder uriBuilder = new URIBuilder(options.getUrl());
-        for (LookupArg arg : params) {
-            uriBuilder.addParameter(arg.getArgName(), arg.getArgValue());
-        }
-
-        return uriBuilder.build();
-    }
-
     private Optional<RowData> processHttpResponse(
             HttpResponse<String> response,
             HttpRequest request) throws IOException {

src/main/java/com/getindata/connectors/http/internal/table/lookup/JavaNetHttpPollingClientFactory.java

@@ -5,6 +5,7 @@
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.table.data.RowData;
 
+import com.getindata.connectors.http.LookupQueryCreator;
 import com.getindata.connectors.http.internal.HeaderPreprocessor;
 import com.getindata.connectors.http.internal.PollingClientFactory;
 import com.getindata.connectors.http.internal.utils.HttpHeaderUtils;
@@ -14,16 +15,22 @@ public class JavaNetHttpPollingClientFactory implements PollingClientFactory<Row
 
     @Override
     public JavaNetHttpPollingClient createPollClient(
-            HttpLookupConfig options,
-            DeserializationSchema<RowData> schemaDecoder) {
+        HttpLookupConfig options,
+        DeserializationSchema<RowData> schemaDecoder,
+        LookupQueryCreator lookupQueryCreator) {
 
         HttpClient httpClient = JavaNetHttpClientFactory.createClient(options.getProperties());
 
         // TODO Consider this to be injected as method argument or factory field
         //  so user could set this using API.
         HeaderPreprocessor headerPreprocessor = HttpHeaderUtils.createDefaultHeaderPreprocessor();
 
-
-        return new JavaNetHttpPollingClient(httpClient, schemaDecoder, options, headerPreprocessor);
+        return new JavaNetHttpPollingClient(
+            httpClient,
+            schemaDecoder,
+            options,
+            lookupQueryCreator,
+            headerPreprocessor
+        );
     }
 }