Improve UDF state handling

If a UDF validation request is in progress while the user continues
editing the definition input, the request should be aborted.

Also keep the state of the UDF editor separate from local storage to
make the data flow in only one direction.

Also changes the UDF validation fetcher to run on empty definitions.

arroyo-console/src/lib/data_fetching.ts

@@ -5,6 +5,7 @@ import { components, paths } from '../gen/api-types';
 import createClient from 'openapi-fetch';
 
 import { LocalUdf } from '../udf_state';
+import { useRef } from 'react';
 
 type schemas = components['schemas'];
 
@@ -121,7 +122,7 @@ const operatorErrorsKey = (pipelineId?: string, jobId?: string) => {
 };
 
 const queryValidationKey = (query?: string, localUdfs?: LocalUdf[]) => {
-  return query ? { key: 'PipelineGraph', query, localUdfs } : null;
+  return query != undefined ? { key: 'PipelineGraph', query, localUdfs } : null;
 };
 
 const udfValidationKey = (definition: string) => {
@@ -372,11 +373,15 @@ export const useQueryValidation = (query?: string, localUdfs?: LocalUdf[]) => {
 };
 
 const udfValidationFetcher = () => {
+  const controller = useRef<AbortController>();
   return async (params: { key: string; definition: string }) => {
+    controller.current?.abort();
+    controller.current = new AbortController();
     const { data, error } = await post('/v1/udfs/validate', {
       body: {
         definition: params.definition,
       },
+      signal: controller.current?.signal,
     });
     return processResponse(data, error);
   };

arroyo-console/src/routes/pipelines/CreatePipeline.tsx

@@ -77,7 +77,7 @@ export function CreatePipeline() {
     useOperatorErrors(pipelineId, job?.id);
   const [operatorErrors, setOperatorErrors] = useState<JobLogMessage[]>([]);
   const [queryInput, setQueryInput] = useState<string>('');
-  const [queryInputToCheck, setQueryInputToCheck] = useState<string>('');
+  const [queryInputToCheck, setQueryInputToCheck] = useState<string | undefined>(undefined);
   const { operatorMetricGroups } = useJobMetrics(pipelineId, job?.id);
   const { isOpen, onOpen, onClose } = useDisclosure();
   const [options, setOptions] = useState<SqlOptions>({ parallelism: 4, checkpointMS: 5000 });

arroyo-console/src/routes/udfs/UdfEditor.tsx

@@ -11,6 +11,7 @@ export interface UdfEditorProps {
 const UdfEditor: React.FC<UdfEditorProps> = ({ udf }) => {
   const { updateLocalUdf, isGlobal } = useContext(LocalUdfsContext);
   const [definitionToCheck, setDefinitionToCheck] = useState<string>(udf.definition);
+  const [localDefinition, setLocalDefinition] = useState<string>(udf.definition);
 
   const debounceSetCheck = useMemo(
     () =>
@@ -30,10 +31,11 @@ const UdfEditor: React.FC<UdfEditorProps> = ({ udf }) => {
 
   return (
     <CodeEditor
-      code={udf.definition}
+      code={localDefinition}
       readOnly={isGlobal(udf)}
       setCode={s => {
         updateLocalUdf(udf as LocalUdf, { definition: s });
+        setLocalDefinition(s);
         debounceSetCheck(s);
       }}
       language="rust"

arroyo-console/src/udf_state.ts

@@ -99,7 +99,6 @@ export const getLocalUdfsContextValue = () => {
   };
 
   const deleteGlobalUdf = async (udf: GlobalUdf) => {
-    console.log('deleting global');
     await apiDeleteGlobalUdf(udf);
     setOpenedGlobalUdfs(openedGlobalUdfs.filter(u => u.id !== udf.id));
   };