🌊 Streams routing UI (elastic#201427)

First iteration of the streams partitioning/routing UI:

<img width="1455" alt="Screenshot 2024-11-27 at 21 31 13"
src="https://github.com/user-attachments/assets/9768b0a0-6143-41a2-9cc0-ed25bbf0d38b">

## Changes

### Unified search bar

An empty `FlexItem` would be rendered even if there is no query input
which will stick around as a 320px wide empty element. This change
avoids rendering this empty wrapper.

### Streams API

* Add logic to extract fields (and their types) from a condition
* Add logic to turn the streams condition dialect into query dsl
* Add dot expander to the root logs pipeline to normalize incoming docs
for consistent access in painless and querydsl (this is a stopgap
solution)
* Add some additional validation to incoming definitions
* Add a sample API which takes a stream and a condition, turns the
condition into querydsl and searches for a bunch of docs and returns
their sources (also sets runtime mappings so non-mapped fields can be
used in the condition)

### Streams app

* Adjust page height based on whether the new nav is enabled
* Add a condition editor to show and change conditions (can also be
reused in other UIs)
* Add UI to show child routing conditions, change existing routings,
partition new streams and delete streams as well

---------

Co-authored-by: Chris Cowan <chris@elastic.co>
Co-authored-by: Dario Gieselaar <dario.gieselaar@elastic.co>
Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>

src/plugins/unified_search/public/query_string_input/query_bar_top_row.tsx

@@ -673,49 +673,59 @@ export const QueryBarTopRow = React.memo(
     }
 
     function renderQueryInput() {
+      const filterButtonGroup = !renderFilterMenuOnly() && renderFilterButtonGroup();
+      const queryInput = shouldRenderQueryInput() && (
+        <EuiFlexItem data-test-subj="unifiedQueryInput">
+          <QueryStringInputUI
+            disableAutoFocus={props.disableAutoFocus}
+            indexPatterns={props.indexPatterns!}
+            query={props.query! as Query}
+            screenTitle={props.screenTitle}
+            onChange={onQueryChange}
+            onChangeQueryInputFocus={onChangeQueryInputFocus}
+            onSubmit={onInputSubmit}
+            persistedLog={persistedLog}
+            dataTestSubj={props.dataTestSubj}
+            placeholder={props.placeholder}
+            isClearable={props.isClearable}
+            iconType={props.iconType}
+            nonKqlMode={props.nonKqlMode}
+            timeRangeForSuggestionsOverride={props.timeRangeForSuggestionsOverride}
+            filtersForSuggestions={props.filtersForSuggestions}
+            disableLanguageSwitcher={true}
+            prepend={renderFilterMenuOnly() && renderFilterButtonGroup()}
+            size={props.suggestionsSize}
+            suggestionsAbstraction={props.suggestionsAbstraction}
+            isDisabled={props.isDisabled}
+            appName={appName}
+            submitOnBlur={props.submitOnBlur}
+            deps={{
+              unifiedSearch,
+              data,
+              storage,
+              usageCollection,
+              notifications,
+              docLinks,
+              http,
+              uiSettings,
+              dataViews,
+            }}
+          />
+        </EuiFlexItem>
+      );
+      if (isQueryLangSelected || (!filterButtonGroup && !queryInput)) {
+        return null;
+      }
       return (
-        <EuiFlexGroup gutterSize="s" responsive={false}>
-          {!renderFilterMenuOnly() && renderFilterButtonGroup()}
-          {shouldRenderQueryInput() && (
-            <EuiFlexItem data-test-subj="unifiedQueryInput">
-              <QueryStringInputUI
-                disableAutoFocus={props.disableAutoFocus}
-                indexPatterns={props.indexPatterns!}
-                query={props.query! as Query}
-                screenTitle={props.screenTitle}
-                onChange={onQueryChange}
-                onChangeQueryInputFocus={onChangeQueryInputFocus}
-                onSubmit={onInputSubmit}
-                persistedLog={persistedLog}
-                dataTestSubj={props.dataTestSubj}
-                placeholder={props.placeholder}
-                isClearable={props.isClearable}
-                iconType={props.iconType}
-                nonKqlMode={props.nonKqlMode}
-                timeRangeForSuggestionsOverride={props.timeRangeForSuggestionsOverride}
-                filtersForSuggestions={props.filtersForSuggestions}
-                disableLanguageSwitcher={true}
-                prepend={renderFilterMenuOnly() && renderFilterButtonGroup()}
-                size={props.suggestionsSize}
-                suggestionsAbstraction={props.suggestionsAbstraction}
-                isDisabled={props.isDisabled}
-                appName={appName}
-                submitOnBlur={props.submitOnBlur}
-                deps={{
-                  unifiedSearch,
-                  data,
-                  storage,
-                  usageCollection,
-                  notifications,
-                  docLinks,
-                  http,
-                  uiSettings,
-                  dataViews,
-                }}
-              />
-            </EuiFlexItem>
-          )}
-        </EuiFlexGroup>
+        <EuiFlexItem
+          grow={!shouldShowDatePickerAsBadge()}
+          style={{ minWidth: shouldShowDatePickerAsBadge() ? 'auto' : 320, maxWidth: '100%' }}
+        >
+          <EuiFlexGroup gutterSize="s" responsive={false}>
+            {filterButtonGroup}
+            {queryInput}
+          </EuiFlexGroup>
+        </EuiFlexItem>
       );
     }
 
@@ -787,12 +797,7 @@ export const QueryBarTopRow = React.memo(
                   adHocDataview={props.indexPatterns?.[0]}
                 />
               )}
-              <EuiFlexItem
-                grow={!shouldShowDatePickerAsBadge()}
-                css={{ minWidth: shouldShowDatePickerAsBadge() ? 'auto' : 320, maxWidth: '100%' }}
-              >
-                {!isQueryLangSelected ? renderQueryInput() : null}
-              </EuiFlexItem>
+              {renderQueryInput()}
               {props.renderQueryInputAppend?.()}
               {shouldShowDatePickerAsBadge() && props.filterBar}
               {renderUpdateButton()}

x-pack/plugins/streams/common/index.ts

@@ -5,4 +5,4 @@
  * 2.0.
  */
 
-export type { StreamDefinition } from './types';
+export type { StreamDefinition, ReadStreamDefinition } from './types';

x-pack/plugins/streams/common/types.ts

@@ -33,11 +33,11 @@ export interface AndCondition {
   and: Condition[];
 }
 
-export interface RerouteOrCondition {
+export interface OrCondition {
   or: Condition[];
 }
 
-export type Condition = FilterCondition | AndCondition | RerouteOrCondition | undefined;
+export type Condition = FilterCondition | AndCondition | OrCondition | undefined;
 
 export const conditionSchema: z.ZodType<Condition> = z.lazy(() =>
   z.union([
@@ -77,17 +77,17 @@ export const fieldDefinitionSchema = z.object({
 
 export type FieldDefinition = z.infer<typeof fieldDefinitionSchema>;
 
+export const streamChildSchema = z.object({
+  id: z.string(),
+  condition: z.optional(conditionSchema),
+});
+
+export type StreamChild = z.infer<typeof streamChildSchema>;
+
 export const streamWithoutIdDefinitonSchema = z.object({
   processing: z.array(processingDefinitionSchema).default([]),
   fields: z.array(fieldDefinitionSchema).default([]),
-  children: z
-    .array(
-      z.object({
-        id: z.string(),
-        condition: z.optional(conditionSchema),
-      })
-    )
-    .default([]),
+  children: z.array(streamChildSchema).default([]),
 });
 
 export type StreamWithoutIdDefinition = z.infer<typeof streamDefinitonSchema>;
@@ -110,3 +110,9 @@ export type StreamDefinition = z.infer<typeof streamDefinitonSchema>;
 export const streamDefinitonWithoutChildrenSchema = streamDefinitonSchema.omit({ children: true });
 
 export type StreamWithoutChildrenDefinition = z.infer<typeof streamDefinitonWithoutChildrenSchema>;
+
+export const readStreamDefinitonSchema = streamDefinitonSchema.extend({
+  inheritedFields: z.array(fieldDefinitionSchema.extend({ from: z.string() })).default([]),
+});
+
+export type ReadStreamDefinition = z.infer<typeof readStreamDefinitonSchema>;

x-pack/plugins/streams/server/lib/streams/helpers/condition_fields.ts

@@ -0,0 +1,85 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { Condition, FilterCondition } from '../../../../common/types';
+import { isAndCondition, isFilterCondition, isOrCondition } from './condition_guards';
+
+export function isComplete(condition: Condition): boolean {
+  if (isFilterCondition(condition)) {
+    return condition.field !== undefined && condition.field !== '';
+  }
+  if (isAndCondition(condition)) {
+    return condition.and.every(isComplete);
+  }
+  if (isOrCondition(condition)) {
+    return condition.or.every(isComplete);
+  }
+  return false;
+}
+
+export function getFields(
+  condition: Condition
+): Array<{ name: string; type: 'number' | 'string' }> {
+  const fields = collectFields(condition);
+  // deduplicate fields, if mapped as string and number, keep as number
+  const uniqueFields = new Map<string, 'number' | 'string'>();
+  fields.forEach((field) => {
+    const existing = uniqueFields.get(field.name);
+    if (existing === 'number') {
+      return;
+    }
+    if (existing === 'string' && field.type === 'number') {
+      uniqueFields.set(field.name, 'number');
+      return;
+    }
+    uniqueFields.set(field.name, field.type);
+  });
+
+  return Array.from(uniqueFields).map(([name, type]) => ({ name, type }));
+}
+
+function collectFields(condition: Condition): Array<{ name: string; type: 'number' | 'string' }> {
+  if (isFilterCondition(condition)) {
+    return [{ name: condition.field, type: getFieldTypeForFilterCondition(condition) }];
+  }
+  if (isAndCondition(condition)) {
+    return condition.and.flatMap(collectFields);
+  }
+  if (isOrCondition(condition)) {
+    return condition.or.flatMap(collectFields);
+  }
+  return [];
+}
+
+function getFieldTypeForFilterCondition(condition: FilterCondition): 'number' | 'string' {
+  switch (condition.operator) {
+    case 'gt':
+    case 'gte':
+    case 'lt':
+    case 'lte':
+      return 'number';
+    case 'neq':
+    case 'eq':
+    case 'exists':
+    case 'contains':
+    case 'startsWith':
+    case 'endsWith':
+    case 'notExists':
+      return 'string';
+    default:
+      return 'string';
+  }
+}
+
+export function validateCondition(condition: Condition) {
+  if (isFilterCondition(condition)) {
+    // check whether a field is specified
+    if (!condition.field.trim()) {
+      throw new Error('Field is required in conditions');
+    }
+  }
+}

x-pack/plugins/streams/server/lib/streams/helpers/condition_guards.ts

@@ -0,0 +1,29 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import {
+  AndCondition,
+  conditionSchema,
+  FilterCondition,
+  filterConditionSchema,
+  OrCondition,
+} from '../../../../common/types';
+
+export function isFilterCondition(subject: any): subject is FilterCondition {
+  const result = filterConditionSchema.safeParse(subject);
+  return result.success;
+}
+
+export function isAndCondition(subject: any): subject is AndCondition {
+  const result = conditionSchema.safeParse(subject);
+  return result.success && subject.and != null;
+}
+
+export function isOrCondition(subject: any): subject is OrCondition {
+  const result = conditionSchema.safeParse(subject);
+  return result.success && subject.or != null;
+}

x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.ts

@@ -7,30 +7,12 @@
 
 import { isBoolean, isString } from 'lodash';
 import {
-  AndCondition,
   BinaryFilterCondition,
   Condition,
-  conditionSchema,
   FilterCondition,
-  filterConditionSchema,
-  RerouteOrCondition,
   UnaryFilterCondition,
 } from '../../../../common/types';
-
-function isFilterCondition(subject: any): subject is FilterCondition {
-  const result = filterConditionSchema.safeParse(subject);
-  return result.success;
-}
-
-function isAndCondition(subject: any): subject is AndCondition {
-  const result = conditionSchema.safeParse(subject);
-  return result.success && subject.and != null;
-}
-
-function isOrCondition(subject: any): subject is RerouteOrCondition {
-  const result = conditionSchema.safeParse(subject);
-  return result.success && subject.or != null;
-}
+import { isAndCondition, isFilterCondition, isOrCondition } from './condition_guards';
 
 function safePainlessField(condition: FilterCondition) {
   return `ctx.${condition.field.split('.').join('?.')}`;

x-pack/plugins/streams/server/lib/streams/helpers/condition_to_query_dsl.ts

@@ -0,0 +1,63 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { Condition, FilterCondition } from '../../../../common/types';
+import { isAndCondition, isFilterCondition, isOrCondition } from './condition_guards';
+
+function conditionToClause(condition: FilterCondition) {
+  switch (condition.operator) {
+    case 'neq':
+      return { bool: { must_not: { match: { [condition.field]: condition.value } } } };
+    case 'eq':
+      return { match: { [condition.field]: condition.value } };
+    case 'exists':
+      return { exists: { field: condition.field } };
+    case 'gt':
+      return { range: { [condition.field]: { gt: condition.value } } };
+    case 'gte':
+      return { range: { [condition.field]: { gte: condition.value } } };
+    case 'lt':
+      return { range: { [condition.field]: { lt: condition.value } } };
+    case 'lte':
+      return { range: { [condition.field]: { lte: condition.value } } };
+    case 'contains':
+      return { wildcard: { [condition.field]: `*${condition.value}*` } };
+    case 'startsWith':
+      return { prefix: { [condition.field]: condition.value } };
+    case 'endsWith':
+      return { wildcard: { [condition.field]: `*${condition.value}` } };
+    case 'notExists':
+      return { bool: { must_not: { exists: { field: condition.field } } } };
+    default:
+      return { match_none: {} };
+  }
+}
+
+export function conditionToQueryDsl(condition: Condition): any {
+  if (isFilterCondition(condition)) {
+    return conditionToClause(condition);
+  }
+  if (isAndCondition(condition)) {
+    const and = condition.and.map((filter) => conditionToQueryDsl(filter));
+    return {
+      bool: {
+        must: and,
+      },
+    };
+  }
+  if (isOrCondition(condition)) {
+    const or = condition.or.map((filter) => conditionToQueryDsl(filter));
+    return {
+      bool: {
+        should: or,
+      },
+    };
+  }
+  return {
+    match_none: {},
+  };
+}

x-pack/plugins/streams/server/lib/streams/ingest_pipelines/logs_default_pipeline.ts

@@ -20,4 +20,9 @@ export const logsDefaultPipelineProcessors = [
       ignore_missing_pipeline: true,
     },
   },
+  {
+    dot_expander: {
+      field: '*',
+    },
+  },
 ];