Added concurrent compaction switches (apache#15114)

* Added concurrent compaction switches
kfaraz · Oct 13, 2023 · 9ca10c7 · 9ca10c7
1 parent 6d62c75
commit 9ca10c7
Show file tree

Hide file tree

Showing 5 changed files with 80 additions and 2 deletions.
diff --git a/web-console/src/components/auto-form/auto-form.tsx b/web-console/src/components/auto-form/auto-form.tsx
@@ -72,6 +72,10 @@ export interface Field<M> {
   hide?: Functor<M, boolean>;
   hideInMore?: Functor<M, boolean>;
   valueAdjustment?: (value: any) => any;
+  /**
+   * An optional callback to transform the value before it is set on the input
+   */
+  adjustValue?: (value: any) => any;
   adjustment?: (model: Partial<M>, oldModel: Partial<M>) => Partial<M>;
   issueWithValue?: (value: any) => string | undefined;
 
@@ -378,12 +382,14 @@ export class AutoForm<T extends Record<string, any>> extends React.PureComponent
     const disabled = AutoForm.evaluateFunctor(field.disabled, model, false);
     const intent = required && modelValue == null ? AutoForm.REQUIRED_INTENT : undefined;
 
+    const adjustedValue = field.adjustValue ? field.adjustValue(shownValue) : shownValue;
+
     return (
       <ButtonGroup large={large}>
         <Button
           intent={intent}
           disabled={disabled}
-          active={shownValue === false}
+          active={adjustedValue === false}
           onClick={() => {
             this.fieldChange(field, false);
             if (onFinalize) onFinalize();
@@ -394,7 +400,7 @@ export class AutoForm<T extends Record<string, any>> extends React.PureComponent
         <Button
           intent={intent}
           disabled={disabled}
-          active={shownValue === true}
+          active={adjustedValue === true}
           onClick={() => {
             this.fieldChange(field, true);
             if (onFinalize) onFinalize();

diff --git a/...src/dialogs/compaction-config-dialog/__snapshots__/compaction-config-dialog.spec.tsx.snap b/...src/dialogs/compaction-config-dialog/__snapshots__/compaction-config-dialog.spec.tsx.snap
@@ -335,6 +335,17 @@ exports[`CompactionConfigDialog matches snapshot with compactionConfig (dynamic
             "name": "tuningConfig.splitHintSpec.maxNumFiles",
             "type": "number",
           },
+          Object {
+            "adjustValue": [Function],
+            "defaultValue": undefined,
+            "info": <p>
+              Allows or forbids concurrent compactions.
+            </p>,
+            "label": "Allow concurrent compactions (experimental)",
+            "name": "taskContext.taskLockType",
+            "type": "boolean",
+            "valueAdjustment": [Function],
+          },
         ]
       }
       model={
@@ -717,6 +728,17 @@ exports[`CompactionConfigDialog matches snapshot with compactionConfig (hashed p
             "name": "tuningConfig.splitHintSpec.maxNumFiles",
             "type": "number",
           },
+          Object {
+            "adjustValue": [Function],
+            "defaultValue": undefined,
+            "info": <p>
+              Allows or forbids concurrent compactions.
+            </p>,
+            "label": "Allow concurrent compactions (experimental)",
+            "name": "taskContext.taskLockType",
+            "type": "boolean",
+            "valueAdjustment": [Function],
+          },
         ]
       }
       model={
@@ -1099,6 +1121,17 @@ exports[`CompactionConfigDialog matches snapshot with compactionConfig (range pa
             "name": "tuningConfig.splitHintSpec.maxNumFiles",
             "type": "number",
           },
+          Object {
+            "adjustValue": [Function],
+            "defaultValue": undefined,
+            "info": <p>
+              Allows or forbids concurrent compactions.
+            </p>,
+            "label": "Allow concurrent compactions (experimental)",
+            "name": "taskContext.taskLockType",
+            "type": "boolean",
+            "valueAdjustment": [Function],
+          },
         ]
       }
       model={
@@ -1481,6 +1514,17 @@ exports[`CompactionConfigDialog matches snapshot without compactionConfig 1`] =
             "name": "tuningConfig.splitHintSpec.maxNumFiles",
             "type": "number",
           },
+          Object {
+            "adjustValue": [Function],
+            "defaultValue": undefined,
+            "info": <p>
+              Allows or forbids concurrent compactions.
+            </p>,
+            "label": "Allow concurrent compactions (experimental)",
+            "name": "taskContext.taskLockType",
+            "type": "boolean",
+            "valueAdjustment": [Function],
+          },
         ]
       }
       model={

diff --git a/web-console/src/druid-models/compaction-config/compaction-config.tsx b/web-console/src/druid-models/compaction-config/compaction-config.tsx
@@ -354,4 +354,13 @@ export const COMPACTION_CONFIG_FIELDS: Field<CompactionConfig>[] = [
       </>
     ),
   },
+  {
+    name: 'taskContext.taskLockType',
+    type: 'boolean',
+    label: 'Allow concurrent compactions (experimental)',
+    defaultValue: undefined,
+    valueAdjustment: v => (v ? 'REPLACE' : undefined),
+    adjustValue: v => v === 'REPLACE',
+    info: <p>Allows or forbids concurrent compactions.</p>,
+  },
 ];
diff --git a/web-console/src/druid-models/ingestion-spec/ingestion-spec.tsx b/web-console/src/druid-models/ingestion-spec/ingestion-spec.tsx
@@ -72,6 +72,7 @@ const CURRENT_YEAR = new Date().getUTCFullYear();
 export interface IngestionSpec {
   readonly type: IngestionType;
   readonly spec: IngestionSpecInner;
+  readonly context?: { taskLockType?: 'APPEND' | 'REPLACE' };
 }
 
 export interface IngestionSpecInner {
@@ -344,6 +345,11 @@ export function normalizeSpec(spec: Partial<IngestionSpec>): IngestionSpec {
   spec = deepSetIfUnset(spec, 'type', specType);
   spec = deepSetIfUnset(spec, 'spec.ioConfig.type', specType);
   spec = deepSetIfUnset(spec, 'spec.tuningConfig.type', specType);
+
+  if (spec.context?.taskLockType !== undefined) {
+    spec.context.taskLockType = spec.spec?.ioConfig.appendToExisting ? 'APPEND' : 'REPLACE';
+  }
+
   return spec as IngestionSpec;
 }
 

diff --git a/web-console/src/views/load-data-view/load-data-view.tsx b/web-console/src/views/load-data-view/load-data-view.tsx
@@ -3134,6 +3134,8 @@ export class LoadDataView extends React.PureComponent<LoadDataViewProps, LoadDat
     const { spec } = this.state;
     const parallel = deepGet(spec, 'spec.tuningConfig.maxNumConcurrentSubTasks') > 1;
 
+    const appendToExisting = spec.spec?.ioConfig.appendToExisting;
+
     return (
       <>
         <div className="main">
@@ -3169,6 +3171,17 @@ export class LoadDataView extends React.PureComponent<LoadDataViewProps, LoadDat
                   </>
                 ),
               },
+              {
+                name: 'context.taskLockType',
+                type: 'boolean',
+                label: `Allow concurrent ${
+                  appendToExisting ? 'append' : 'replace'
+                } tasks (experimental)`,
+                defaultValue: undefined,
+                valueAdjustment: v => (v ? (appendToExisting ? 'APPEND' : 'REPLACE') : undefined),
+                adjustValue: v => v === (appendToExisting ? 'APPEND' : 'REPLACE'),
+                info: <p>Allows or forbids concurrent tasks.</p>,
+              },
             ]}
             model={spec}
             onChange={this.updateSpec}