onboard normalization processor (#311)

Signed-off-by: Tyler Ohlsen <[email protected]>
(cherry picked from commit e913e41)

common/constants.ts

@@ -73,6 +73,7 @@ export enum PROCESSOR_TYPE {
   SPLIT = 'split',
   SORT = 'sort',
   TEXT_CHUNKING = 'text_chunking',
+  NORMALIZATION = 'normalization-processor',
 }
 
 export enum MODEL_TYPE {
@@ -129,6 +130,8 @@ export const TEXT_CHUNKING_PROCESSOR_LINK =
   'https://opensearch.org/docs/latest/ingest-pipelines/processors/text-chunking/';
 export const CREATE_WORKFLOW_LINK =
   'https://opensearch.org/docs/latest/automating-configurations/api/create-workflow/';
+export const NORMALIZATION_PROCESSOR_LINK =
+  'https://opensearch.org/docs/latest/search-plugins/search-pipelines/normalization-processor/';
 
 /**
  * Text chunking algorithm constants

public/configs/search_response_processors/index.ts

@@ -6,3 +6,4 @@
 export * from './ml_search_response_processor';
 export * from './split_search_response_processor';
 export * from './sort_search_response_processor';
+export * from './normalization_processor';

public/configs/search_response_processors/normalization_processor.ts

@@ -0,0 +1,43 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { PROCESSOR_TYPE } from '../../../common';
+import { Processor } from '../processor';
+
+/**
+ * The normalization processor config. Used in search flows.
+ */
+export class NormalizationProcessor extends Processor {
+  constructor() {
+    super();
+    this.type = PROCESSOR_TYPE.NORMALIZATION;
+    this.name = 'Normalization Processor';
+    this.fields = [];
+    this.optionalFields = [
+      {
+        id: 'weights',
+        type: 'string',
+      },
+      {
+        id: 'normalization_technique',
+        type: 'select',
+        selectOptions: ['min_max', 'l2'],
+      },
+      {
+        id: 'combination_technique',
+        type: 'select',
+        selectOptions: ['arithmetic_mean', 'geometric_mean', 'harmonic_mean'],
+      },
+      {
+        id: 'description',
+        type: 'string',
+      },
+      {
+        id: 'tag',
+        type: 'string',
+      },
+    ];
+  }
+}

public/pages/workflow_detail/workflow_inputs/processor_inputs/normalization_processor_inputs.tsx

@@ -0,0 +1,65 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import React from 'react';
+import { EuiAccordion, EuiFlexItem, EuiSpacer } from '@elastic/eui';
+import {
+  IProcessorConfig,
+  PROCESSOR_CONTEXT,
+  WorkflowConfig,
+  NORMALIZATION_PROCESSOR_LINK,
+} from '../../../../../common';
+import { TextField } from '../input_fields';
+import { ConfigFieldList } from '../config_field_list';
+
+interface NormalizationProcessorInputsProps {
+  uiConfig: WorkflowConfig;
+  config: IProcessorConfig;
+  baseConfigPath: string; // the base path of the nested config, if applicable. e.g., 'ingest.enrich'
+  context: PROCESSOR_CONTEXT;
+}
+
+/**
+ * Specialized component to render the normalization processor. Adds some helper text around weights field.
+ * In the future, may have a more customizable / guided way for specifying the array of weights.
+ * For example, could have some visual way of linking it to the underlying sub-queries in the query field,
+ * enforce its length = the number of queries, etc.
+ */
+export function NormalizationProcessorInputs(
+  props: NormalizationProcessorInputsProps
+) {
+  // extracting field info from the config
+  const optionalFields = props.config.optionalFields || [];
+  const weightsFieldPath = `${props.baseConfigPath}.${props.config.id}.weights`;
+  const optionalFieldsWithoutWeights = optionalFields.filter(
+    (field) => field.id !== 'weights'
+  );
+
+  return (
+    // We only have optional fields for this processor, so everything is nested under the accordion
+    <EuiAccordion
+      id={`advancedSettings${props.config.id}`}
+      buttonContent="Advanced settings"
+      paddingSize="none"
+    >
+      <EuiSpacer size="s" />
+      <EuiFlexItem>
+        <TextField
+          label={'Weights'}
+          helpText={`A comma-separated array of floating-point values specifying the weight for each query. For example: '0.8, 0.2'`}
+          helpLink={NORMALIZATION_PROCESSOR_LINK}
+          fieldPath={weightsFieldPath}
+          showError={true}
+        />
+      </EuiFlexItem>
+      <EuiSpacer size="s" />
+      <ConfigFieldList
+        configId={props.config.id}
+        configFields={optionalFieldsWithoutWeights}
+        baseConfigPath={props.baseConfigPath}
+      />
+    </EuiAccordion>
+  );
+}

public/pages/workflow_detail/workflow_inputs/processor_inputs/processor_inputs.tsx

@@ -15,6 +15,7 @@ import {
 import { MLProcessorInputs } from './ml_processor_inputs';
 import { ConfigFieldList } from '../config_field_list';
 import { TextChunkingProcessorInputs } from './text_chunking_processor_inputs';
+import { NormalizationProcessorInputs } from './normalization_processor_inputs';
 
 /**
  * Base component for rendering processor form inputs based on the processor type
@@ -69,6 +70,20 @@ export function ProcessorInputs(props: ProcessorInputsProps) {
             );
             break;
           }
+          case PROCESSOR_TYPE.NORMALIZATION: {
+            el = (
+              <EuiFlexItem>
+                <NormalizationProcessorInputs
+                  uiConfig={props.uiConfig}
+                  config={props.config}
+                  baseConfigPath={props.baseConfigPath}
+                  context={props.context}
+                />
+                <EuiSpacer size={PROCESSOR_INPUTS_SPACER_SIZE} />
+              </EuiFlexItem>
+            );
+            break;
+          }
           default: {
             el = (
               <EuiFlexItem>

public/pages/workflow_detail/workflow_inputs/processors_list.tsx

@@ -28,6 +28,7 @@ import {
   MLIngestProcessor,
   MLSearchRequestProcessor,
   MLSearchResponseProcessor,
+  NormalizationProcessor,
   SortIngestProcessor,
   SortSearchResponseProcessor,
   SplitIngestProcessor,
@@ -272,6 +273,15 @@ export function ProcessorsList(props: ProcessorsListProps) {
                               );
                             },
                           },
+                          {
+                            name: 'Normalization Processor',
+                            onClick: () => {
+                              closePopover();
+                              addProcessor(
+                                new NormalizationProcessor().toObj()
+                              );
+                            },
+                          },
                         ],
                 },
               ]}

public/pages/workflows/new_workflow/quick_configure_inputs.tsx

@@ -65,7 +65,7 @@ export function QuickConfigureInputs(props: QuickConfigureInputsProps) {
           <EuiAccordion
             id="optionalConfiguration"
             buttonContent="Optional configuration"
-            initialIsOpen={true}
+            initialIsOpen={false}
           >
             <EuiSpacer size="m" />
             <EuiCompressedFormRow

public/utils/config_to_template_utils.ts

@@ -116,11 +116,26 @@ function searchConfigToTemplateNodes(
   const searchRequestProcessors = processorConfigsToTemplateProcessors(
     searchConfig.enrichRequest.processors
   );
+  // For the configured response processors, we don't maintain separate UI / config
+  // between response processors and phase results processors. So, we parse
+  // out those different processor types here when configuring the final search pipeline.
+  // Currently, the only special phase results processor supported is the normalization processor,
+  // so we filter & partition on that type.
+  const normalizationProcessor = searchConfig.enrichResponse.processors.find(
+    (processor) => processor.type === PROCESSOR_TYPE.NORMALIZATION
+  );
+  const phaseResultsProcessors = processorConfigsToTemplateProcessors(
+    normalizationProcessor ? [normalizationProcessor] : []
+  );
   const searchResponseProcessors = processorConfigsToTemplateProcessors(
-    searchConfig.enrichResponse.processors
+    searchConfig.enrichResponse.processors.filter(
+      (processor) => processor.type !== PROCESSOR_TYPE.NORMALIZATION
+    )
   );
   const hasProcessors =
-    searchRequestProcessors.length > 0 || searchResponseProcessors.length > 0;
+    searchRequestProcessors.length > 0 ||
+    searchResponseProcessors.length > 0 ||
+    phaseResultsProcessors.length > 0;
 
   return hasProcessors
     ? [
@@ -133,6 +148,7 @@ function searchConfigToTemplateNodes(
             configurations: JSON.stringify({
               request_processors: searchRequestProcessors,
               response_processors: searchResponseProcessors,
+              phase_results_processors: phaseResultsProcessors,
             } as SearchPipelineConfig),
           },
         } as CreateSearchPipelineNode,
@@ -254,6 +270,49 @@ export function processorConfigsToTemplateProcessors(
         });
         break;
       }
+      // optionally add any parameters specified, in the expected nested format
+      // for the normalization processor
+      case PROCESSOR_TYPE.NORMALIZATION: {
+        const {
+          normalization_technique,
+          combination_technique,
+          weights,
+        } = processorConfigToFormik(processorConfig);
+
+        let finalConfig = {} as any;
+        if (!isEmpty(normalization_technique)) {
+          finalConfig = {
+            ...finalConfig,
+            normalization: {
+              technique: normalization_technique,
+            },
+          };
+        }
+        if (!isEmpty(combination_technique)) {
+          finalConfig = {
+            ...finalConfig,
+            combination: {
+              ...(finalConfig?.combination || {}),
+              technique: combination_technique,
+            },
+          };
+        }
+        if (!isEmpty(weights) && weights.split(',').length > 0) {
+          finalConfig = {
+            ...finalConfig,
+            combination: {
+              ...(finalConfig?.combination || {}),
+              parameters: {
+                weights: weights.split(',').map(Number),
+              },
+            },
+          };
+        }
+        processorsList.push({
+          [processorConfig.type]: finalConfig,
+        });
+        break;
+      }
       case PROCESSOR_TYPE.SPLIT:
       case PROCESSOR_TYPE.SORT:
       default: {