add watermark and datasource empty check

Signed-off-by: Shenoy Pratik <[email protected]>
ps48 · Oct 24, 2023 · f1e88ed · f1e88ed
1 parent f614db4
commit f1e88ed
Show file tree

Hide file tree

Showing 5 changed files with 112 additions and 22 deletions.
diff --git a/common/types/index.ts b/common/types/index.ts
@@ -32,6 +32,11 @@ export interface RefreshIntervalType {
   refreshInterval: string;
 }
 
+export interface watermarkDelayType {
+  delayWindow: number;
+  delayInterval: string;
+}
+
 export type AccelerationIndexType = 'skipping' | 'covering' | 'materialized';
 
 export interface GroupByTumbleType {
@@ -57,6 +62,7 @@ export interface FormErrorsType {
   replicaShardsError: string[];
   refreshIntervalError: string[];
   checkpointLocationError: string[];
+  watermarkDelayError: string[];
 }
 
 export type AccelerationRefreshType = 'auto' | 'interval' | 'manual';
@@ -75,12 +81,20 @@ export interface CreateAccelerationForm {
   replicaShardsCount: number;
   refreshType: AccelerationRefreshType;
   checkpointLocation: string | undefined;
+  watermarkDelay: watermarkDelayType;
   refreshIntervalOptions: RefreshIntervalType;
   formErrors: FormErrorsType;
 }
 
 export type AsyncQueryLoadingStatus = 'SUCCESS' | 'FAILED' | 'RUNNING' | 'SCHEDULED' | 'CANCELLED';
-export type TreeItemType = 'covering_index' | 'skipping_index' | 'table' | 'database' | 'materialized_view' | 'Load Materialized View' | 'badge'
+export type TreeItemType =
+  | 'covering_index'
+  | 'skipping_index'
+  | 'table'
+  | 'database'
+  | 'materialized_view'
+  | 'Load Materialized View'
+  | 'badge';
 
 export interface TreeItem {
   name: string;

diff --git a/public/components/acceleration/create/create_acceleration.tsx b/public/components/acceleration/create/create_acceleration.tsx
@@ -65,6 +65,10 @@ export const CreateAcceleration = ({
     replicaShardsCount: 1,
     refreshType: 'auto',
     checkpointLocation: undefined,
+    watermarkDelay: {
+      delayWindow: 1,
+      delayInterval: ACCELERATION_TIME_INTERVAL[1].value,
+    },
     refreshIntervalOptions: {
       refreshWindow: 1,
       refreshInterval: ACCELERATION_TIME_INTERVAL[1].value,
@@ -81,6 +85,7 @@ export const CreateAcceleration = ({
       replicaShardsError: [],
       refreshIntervalError: [],
       checkpointLocationError: [],
+      watermarkDelayError: [],
     },
   });
 

diff --git a/public/components/acceleration/create/utils.tsx b/public/components/acceleration/create/utils.tsx
@@ -51,6 +51,15 @@ export const validateRefreshInterval = (refreshType: string, refreshWindow: numb
     : [];
 };
 
+export const validateWatermarkDelay = (
+  accelerationIndexType: AccelerationIndexType,
+  delayWindow: number
+) => {
+  return accelerationIndexType === 'materialized' && delayWindow < 1
+    ? ['delay window should be greater than 0']
+    : [];
+};
+
 export const validateIndexName = (value: string) => {
   // Check if the value does not begin with underscores or hyphens and all characters are lower case
   return !ACCELERATION_INDEX_NAME_REGEX.test(value) ? ['Enter a valid index name'] : [];
@@ -123,10 +132,11 @@ export const formValidator = (accelerationformData: CreateAccelerationForm) => {
     replicaShardsCount,
     refreshType,
     checkpointLocation,
+    watermarkDelay,
     refreshIntervalOptions,
   } = accelerationformData;
 
-  const accelerationFormErrors = {
+  const accelerationFormErrors: FormErrorsType = {
     dataSourceError: validateDataSource(dataSource),
     databaseError: validateDatabase(database),
     dataTableError: validateDataTable(dataTable),
@@ -137,6 +147,7 @@ export const formValidator = (accelerationformData: CreateAccelerationForm) => {
       refreshIntervalOptions.refreshWindow
     ),
     checkpointLocationError: validateCheckpointLocation(refreshType, checkpointLocation),
+    watermarkDelayError: validateWatermarkDelay(accelerationIndexType, watermarkDelay.delayWindow),
     indexNameError: validateIndexName(accelerationIndexName),
     skippingIndexError: validateSkippingIndexData(accelerationIndexType, skippingIndexQueryData),
     coveringIndexError: validateCoveringIndexData(accelerationIndexType, coveringIndexQueryData),

diff --git a/public/components/acceleration/selectors/index_setting_options.tsx b/public/components/acceleration/selectors/index_setting_options.tsx
@@ -23,6 +23,7 @@ import {
   validatePrimaryShardCount,
   validateRefreshInterval,
   validateReplicaCount,
+  validateWatermarkDelay,
 } from '../create/utils';
 import { IndexTypeSelector } from './index_type_selector';
 
@@ -60,6 +61,8 @@ export const IndexSettingOptions = ({
   const [refreshTypeSelected, setRefreshTypeSelected] = useState(autoRefreshId);
   const [refreshWindow, setRefreshWindow] = useState(1);
   const [refreshInterval, setRefreshInterval] = useState(ACCELERATION_TIME_INTERVAL[1].value);
+  const [delayWindow, setDelayWindow] = useState(1);
+  const [delayInterval, setDelayInterval] = useState(ACCELERATION_TIME_INTERVAL[1].value);
   const [checkpoint, setCheckpoint] = useState('');
 
   const onChangePrimaryShards = (e: ChangeEvent<HTMLInputElement>) => {
@@ -104,6 +107,16 @@ export const IndexSettingOptions = ({
     setRefreshWindow(windowCount);
   };
 
+  const onChangeDelayWindow = (e: ChangeEvent<HTMLInputElement>) => {
+    const windowCount = parseInt(e.target.value, 10);
+    setAccelerationFormData(
+      producer((accData) => {
+        accData.watermarkDelay.delayWindow = windowCount;
+      })
+    );
+    setDelayWindow(windowCount);
+  };
+
   const onChangeRefreshInterval = (e: React.ChangeEvent<HTMLSelectElement>) => {
     const refreshIntervalValue = e.target.value;
     setAccelerationFormData(
@@ -114,6 +127,16 @@ export const IndexSettingOptions = ({
     setRefreshInterval(refreshIntervalValue);
   };
 
+  const onChangeDelayInterval = (e: React.ChangeEvent<HTMLSelectElement>) => {
+    const delayIntervalValue = e.target.value;
+    setAccelerationFormData(
+      producer((accData) => {
+        accData.watermarkDelay.delayInterval = delayIntervalValue;
+      })
+    );
+    setRefreshInterval(delayIntervalValue);
+  };
+
   const onChangeCheckpoint = (e: ChangeEvent<HTMLInputElement>) => {
     const checkpointLocation = e.target.value;
     setAccelerationFormData({ ...accelerationFormData, checkpointLocation: checkpointLocation });
@@ -252,6 +275,40 @@ export const IndexSettingOptions = ({
           />
         </EuiFormRow>
       )}
+      {accelerationFormData.accelerationIndexType === 'materialized' && (
+        <EuiFormRow
+          label="Watermark Delay"
+          helpText="Data arrival delay for incremental refresh on a materialized view with aggregations"
+          isInvalid={hasError(accelerationFormData.formErrors, 'watermarkDelayError')}
+          error={accelerationFormData.formErrors.watermarkDelayError}
+        >
+          <EuiFieldNumber
+            placeholder="Watermark delay interval"
+            value={delayWindow}
+            onChange={onChangeDelayWindow}
+            aria-label="Watermark delay interval"
+            min={1}
+            isInvalid={hasError(accelerationFormData.formErrors, 'watermarkDelayError')}
+            onBlur={(e) => {
+              setAccelerationFormData(
+                producer((accData) => {
+                  accData.formErrors.watermarkDelayError = validateWatermarkDelay(
+                    accelerationFormData.accelerationIndexType,
+                    parseInt(e.target.value, 10)
+                  );
+                })
+              );
+            }}
+            append={
+              <EuiSelect
+                value={delayInterval}
+                onChange={onChangeDelayInterval}
+                options={ACCELERATION_TIME_INTERVAL}
+              />
+            }
+          />
+        </EuiFormRow>
+      )}
     </>
   );
 };
diff --git a/public/components/acceleration/selectors/source_selector.tsx b/public/components/acceleration/selectors/source_selector.tsx
@@ -129,14 +129,15 @@ export const AccelerationDataSourceSelector = ({
           options={dataConnections}
           selectedOptions={selectedDataConnection}
           onChange={(dataConnectionOptions) => {
-            setAccelerationFormData(
-              producer((accData) => {
-                accData.dataSource = dataConnectionOptions[0].label;
-                accData.formErrors.dataSourceError = validateDataSource(
-                  dataConnectionOptions[0].label
-                );
-              })
-            );
+            if (dataConnectionOptions.length > 0)
+              setAccelerationFormData(
+                producer((accData) => {
+                  accData.dataSource = dataConnectionOptions[0].label;
+                  accData.formErrors.dataSourceError = validateDataSource(
+                    dataConnectionOptions[0].label
+                  );
+                })
+              );
             setSelectedDataConnection(dataConnectionOptions);
           }}
           isClearable={false}
@@ -156,12 +157,13 @@ export const AccelerationDataSourceSelector = ({
           options={databases}
           selectedOptions={selectedDatabase}
           onChange={(databaseOptions) => {
-            setAccelerationFormData(
-              producer((accData) => {
-                accData.database = databaseOptions[0].label;
-                accData.formErrors.databaseError = validateDataSource(databaseOptions[0].label);
-              })
-            );
+            if (databaseOptions.length > 0)
+              setAccelerationFormData(
+                producer((accData) => {
+                  accData.database = databaseOptions[0].label;
+                  accData.formErrors.databaseError = validateDataSource(databaseOptions[0].label);
+                })
+              );
             setSelectedDatabase(databaseOptions);
           }}
           isClearable={false}
@@ -181,12 +183,13 @@ export const AccelerationDataSourceSelector = ({
           options={tables}
           selectedOptions={selectedTable}
           onChange={(tableOptions) => {
-            setAccelerationFormData(
-              producer((accData) => {
-                accData.dataTable = tableOptions[0].label;
-                accData.formErrors.dataTableError = validateDataSource(tableOptions[0].label);
-              })
-            );
+            if (tableOptions.length > 0)
+              setAccelerationFormData(
+                producer((accData) => {
+                  accData.dataTable = tableOptions[0].label;
+                  accData.formErrors.dataTableError = validateDataSource(tableOptions[0].label);
+                })
+              );
             setSelectedTable(tableOptions);
           }}
           isClearable={false}