Upstream: 9dfb37871162d884505ac4f012270502644adeb6

docs/api-reference/dynamic-configuration-api.md

@@ -91,8 +91,6 @@ Host: http://ROUTER_IP:ROUTER_PORT
 ```json
 {
     "millisToWaitBeforeDeleting": 900000,
-    "mergeBytesLimit": 524288000,
-    "mergeSegmentsLimit": 100,
     "maxSegmentsToMove": 100,
     "replicantLifetime": 15,
     "replicationThrottleLimit": 500,
@@ -159,8 +157,6 @@ curl "http://ROUTER_IP:ROUTER_PORT/druid/coordinator/v1/config" \
 --header 'Content-Type: application/json' \
 --data '{
   "millisToWaitBeforeDeleting": 900000,
-  "mergeBytesLimit": 524288000,
-  "mergeSegmentsLimit": 100,
   "maxSegmentsToMove": 5,
   "percentOfSegmentsToConsiderPerMove": 100,
   "useBatchedSegmentSampler": true,
@@ -192,8 +188,6 @@ Content-Length: 683
 
 {
   "millisToWaitBeforeDeleting": 900000,
-  "mergeBytesLimit": 524288000,
-  "mergeSegmentsLimit": 100,
   "maxSegmentsToMove": 5,
   "percentOfSegmentsToConsiderPerMove": 100,
   "useBatchedSegmentSampler": true,
@@ -295,7 +289,7 @@ Host: http://ROUTER_IP:ROUTER_PORT
             "comment": "",
             "ip": "127.0.0.1"
         },
-        "payload": "{\"millisToWaitBeforeDeleting\":900000,\"mergeBytesLimit\":524288000,\"mergeSegmentsLimit\":100,\"maxSegmentsToMove\":5,\"replicantLifetime\":15,\"replicationThrottleLimit\":10,\"balancerComputeThreads\":1,\"killDataSourceWhitelist\":[],\"killPendingSegmentsSkipList\":[],\"maxSegmentsInNodeLoadingQueue\":100,\"decommissioningNodes\":[],\"decommissioningMaxPercentOfMaxSegmentsToMove\":70,\"pauseCoordination\":false,\"replicateAfterLoadTimeout\":false,\"maxNonPrimaryReplicantsToLoad\":2147483647,\"useRoundRobinSegmentAssignment\":true,\"smartSegmentLoading\":true,\"debugDimensions\":null}",
+        "payload": "{\"millisToWaitBeforeDeleting\":900000,\"maxSegmentsToMove\":5,\"replicantLifetime\":15,\"replicationThrottleLimit\":10,\"balancerComputeThreads\":1,\"killDataSourceWhitelist\":[],\"killPendingSegmentsSkipList\":[],\"maxSegmentsInNodeLoadingQueue\":100,\"decommissioningNodes\":[],\"decommissioningMaxPercentOfMaxSegmentsToMove\":70,\"pauseCoordination\":false,\"replicateAfterLoadTimeout\":false,\"maxNonPrimaryReplicantsToLoad\":2147483647,\"useRoundRobinSegmentAssignment\":true,\"smartSegmentLoading\":true,\"debugDimensions\":null}",
         "auditTime": "2023-10-03T20:59:51.622Z"
     }
 ]

docs/configuration/index.md

@@ -951,8 +951,6 @@ The following table shows the dynamic configuration properties for the Coordinat
 |Property|Description|Default|
 |--------|-----------|-------|
 |`millisToWaitBeforeDeleting`|How long does the Coordinator need to be a leader before it can start marking overshadowed segments as unused in metadata storage.| 900000 (15 mins)|
-|`mergeBytesLimit`|The maximum total uncompressed size in bytes of segments to merge.|524288000L|
-|`mergeSegmentsLimit`|The maximum number of segments that can be in a single [append task](../ingestion/tasks.md).|100|
 |`smartSegmentLoading`|Enables ["smart" segment loading mode](#smart-segment-loading) which dynamically computes the optimal values of several properties that maximize Coordinator performance.|true|
 |`maxSegmentsToMove`|The maximum number of segments that can be moved in a Historical tier at any given time.|100|
 |`replicantLifetime`|The maximum number of Coordinator runs for which a segment can wait in the load queue of a Historical before Druid raises an alert.|15|

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/MSQCompactionRunner.java

@@ -29,6 +29,7 @@
 import com.google.inject.Injector;
 import org.apache.druid.client.indexing.ClientCompactionRunnerInfo;
 import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.data.input.impl.DimensionsSpec;
 import org.apache.druid.indexer.TaskStatus;
 import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
 import org.apache.druid.indexer.partitions.PartitionsSpec;
@@ -84,6 +85,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -130,7 +132,7 @@ public MSQCompactionRunner(
    * The following configs aren't supported:
    * <ul>
    * <li>partitionsSpec of type HashedParititionsSpec.</li>
-   * <li>'range' partitionsSpec with non-string partition dimensions.</li>
+   * <li>'range' partitionsSpec with multi-valued or non-string partition dimensions.</li>
    * <li>maxTotalRows in DynamicPartitionsSpec.</li>
    * <li>Rollup without metricsSpec being specified or vice-versa.</li>
    * <li>Any aggregatorFactory {@code A} s.t. {@code A != A.combiningFactory()}.</li>
@@ -153,13 +155,24 @@ public CompactionConfigValidationResult validateCompactionTask(
       );
     }
     List<CompactionConfigValidationResult> validationResults = new ArrayList<>();
+    DataSchema dataSchema = Iterables.getOnlyElement(intervalToDataSchemaMap.values());
     if (compactionTask.getTuningConfig() != null) {
       validationResults.add(
           ClientCompactionRunnerInfo.validatePartitionsSpecForMSQ(
               compactionTask.getTuningConfig().getPartitionsSpec(),
-              Iterables.getOnlyElement(intervalToDataSchemaMap.values()).getDimensionsSpec().getDimensions()
+              dataSchema.getDimensionsSpec().getDimensions()
           )
       );
+      validationResults.add(
+          validatePartitionDimensionsAreNotMultiValued(
+              compactionTask.getTuningConfig().getPartitionsSpec(),
+              dataSchema.getDimensionsSpec(),
+              dataSchema instanceof CombinedDataSchema
+              ? ((CombinedDataSchema) dataSchema).getMultiValuedDimensions()
+              : null
+          )
+      );
+
     }
     if (compactionTask.getGranularitySpec() != null) {
       validationResults.add(ClientCompactionRunnerInfo.validateRollupForMSQ(
@@ -175,6 +188,32 @@ public CompactionConfigValidationResult validateCompactionTask(
                             .orElse(CompactionConfigValidationResult.success());
   }
 
+  private CompactionConfigValidationResult validatePartitionDimensionsAreNotMultiValued(
+      PartitionsSpec partitionsSpec,
+      DimensionsSpec dimensionsSpec,
+      Set<String> multiValuedDimensions
+  )
+  {
+    List<String> dimensionSchemas = dimensionsSpec.getDimensionNames();
+    if (partitionsSpec instanceof DimensionRangePartitionsSpec
+        && dimensionSchemas != null
+        && multiValuedDimensions != null
+        && !multiValuedDimensions.isEmpty()) {
+      Optional<String> multiValuedDimension = ((DimensionRangePartitionsSpec) partitionsSpec)
+          .getPartitionDimensions()
+          .stream()
+          .filter(multiValuedDimensions::contains)
+          .findAny();
+      if (multiValuedDimension.isPresent()) {
+        return CompactionConfigValidationResult.failure(
+            "MSQ: Multi-valued string partition dimension[%s] not supported with 'range' partition spec",
+            multiValuedDimension.get()
+        );
+      }
+    }
+    return CompactionConfigValidationResult.success();
+  }
+
   @Override
   public CurrentSubTaskHolder getCurrentSubTaskHolder()
   {

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/TaskStartTimeoutFault.java

@@ -45,7 +45,7 @@ public TaskStartTimeoutFault(
   {
     super(
         CODE,
-        "Unable to launch [%d] workers out of the total [%d] worker tasks within [%,d] seconds of the last successful worker launch."
+        "Unable to launch [%d] workers out of the total [%d] worker tasks within [%,d] seconds of the last successful worker launch. "
         + "There might be insufficient available slots to start all worker tasks simultaneously. "
         + "Try lowering '%s' in your query context to a number that fits within your available task capacity, "
         + "or try increasing capacity.",

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/resources/SqlTaskResource.java

@@ -33,7 +33,6 @@
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.msq.guice.MultiStageQuery;
 import org.apache.druid.msq.sql.MSQTaskSqlEngine;
-import org.apache.druid.msq.sql.SqlTaskStatus;
 import org.apache.druid.query.QueryException;
 import org.apache.druid.server.QueryResponse;
 import org.apache.druid.server.initialization.ServerConfig;
@@ -48,6 +47,7 @@
 import org.apache.druid.sql.http.ResultFormat;
 import org.apache.druid.sql.http.SqlQuery;
 import org.apache.druid.sql.http.SqlResource;
+import org.apache.druid.sql.http.SqlTaskStatus;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.Consumes;

extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java

@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 import org.apache.druid.client.indexing.ClientCompactionTaskGranularitySpec;
 import org.apache.druid.client.indexing.ClientCompactionTaskTransformSpec;
@@ -109,11 +110,15 @@ public class MSQCompactionRunnerTest
   );
   private static final Map<Interval, DataSchema> INTERVAL_DATASCHEMAS = ImmutableMap.of(
       COMPACTION_INTERVAL,
-      new DataSchema.Builder()
-          .withDataSource(DATA_SOURCE)
-          .withTimestamp(new TimestampSpec(TIMESTAMP_COLUMN, null, null))
-          .withDimensions(new DimensionsSpec(DIMENSIONS))
-          .build()
+      new CombinedDataSchema(
+          DATA_SOURCE,
+          new TimestampSpec(TIMESTAMP_COLUMN, null, null),
+          new DimensionsSpec(DIMENSIONS),
+          null,
+          null,
+          null,
+          ImmutableSet.of(MV_STRING_DIMENSION.getName())
+      )
   );
   private static final ObjectMapper JSON_MAPPER = new DefaultObjectMapper();
   private static final AggregatorFactory AGG1 = new CountAggregatorFactory("agg_0");
@@ -139,6 +144,7 @@ public void testMultipleDisjointCompactionIntervalsAreInvalid()
         null,
         Collections.emptyMap(),
         null,
+        null,
         null
     );
     CompactionConfigValidationResult validationResult = MSQ_COMPACTION_RUNNER.validateCompactionTask(
@@ -160,6 +166,7 @@ public void testHashedPartitionsSpecIsInvalid()
         null,
         Collections.emptyMap(),
         null,
+        null,
         null
     );
     Assert.assertFalse(MSQ_COMPACTION_RUNNER.validateCompactionTask(compactionTask, INTERVAL_DATASCHEMAS).isValid());
@@ -173,6 +180,7 @@ public void testStringDimensionInRangePartitionsSpecIsValid()
         null,
         Collections.emptyMap(),
         null,
+        null,
         null
     );
     Assert.assertTrue(MSQ_COMPACTION_RUNNER.validateCompactionTask(compactionTask, INTERVAL_DATASCHEMAS).isValid());
@@ -187,6 +195,7 @@ public void testLongDimensionInRangePartitionsSpecIsInvalid()
         null,
         Collections.emptyMap(),
         null,
+        null,
         null
     );
 
@@ -200,6 +209,29 @@ public void testLongDimensionInRangePartitionsSpecIsInvalid()
     );
   }
 
+  @Test
+  public void testMultiValuedDimensionInRangePartitionsSpecIsInvalid()
+  {
+    List<String> mvStringPartitionDimension = Collections.singletonList(MV_STRING_DIMENSION.getName());
+    CompactionTask compactionTask = createCompactionTask(
+        new DimensionRangePartitionsSpec(TARGET_ROWS_PER_SEGMENT, null, mvStringPartitionDimension, false),
+        null,
+        Collections.emptyMap(),
+        null,
+        null,
+        null
+    );
+
+    CompactionConfigValidationResult validationResult = MSQ_COMPACTION_RUNNER.validateCompactionTask(compactionTask,
+                                                                                                     INTERVAL_DATASCHEMAS
+    );
+    Assert.assertFalse(validationResult.isValid());
+    Assert.assertEquals(
+        "MSQ: Multi-valued string partition dimension[mv_string_dim] not supported with 'range' partition spec",
+        validationResult.getReason()
+    );
+  }
+
   @Test
   public void testMaxTotalRowsIsInvalid()
   {
@@ -208,6 +240,7 @@ public void testMaxTotalRowsIsInvalid()
         null,
         Collections.emptyMap(),
         null,
+        null,
         null
     );
     Assert.assertFalse(MSQ_COMPACTION_RUNNER.validateCompactionTask(compactionTask, INTERVAL_DATASCHEMAS).isValid());
@@ -221,6 +254,7 @@ public void testDynamicPartitionsSpecIsValid()
         null,
         Collections.emptyMap(),
         null,
+        null,
         null
     );
     Assert.assertTrue(MSQ_COMPACTION_RUNNER.validateCompactionTask(compactionTask, INTERVAL_DATASCHEMAS).isValid());
@@ -234,6 +268,7 @@ public void testQueryGranularityAllIsValid()
         null,
         Collections.emptyMap(),
         new ClientCompactionTaskGranularitySpec(null, Granularities.ALL, null),
+        null,
         null
     );
     Assert.assertTrue(MSQ_COMPACTION_RUNNER.validateCompactionTask(compactionTask, INTERVAL_DATASCHEMAS).isValid());
@@ -247,26 +282,28 @@ public void testRollupFalseWithMetricsSpecIsInValid()
         null,
         Collections.emptyMap(),
         new ClientCompactionTaskGranularitySpec(null, null, false),
+        null,
         AGGREGATORS.toArray(new AggregatorFactory[0])
     );
     Assert.assertFalse(MSQ_COMPACTION_RUNNER.validateCompactionTask(compactionTask, INTERVAL_DATASCHEMAS).isValid());
   }
 
   @Test
-  public void testRollupTrueWithoutMetricsSpecIsInValid()
+  public void testRollupTrueWithoutMetricsSpecIsInvalid()
   {
     CompactionTask compactionTask = createCompactionTask(
         new DynamicPartitionsSpec(3, null),
         null,
         Collections.emptyMap(),
         new ClientCompactionTaskGranularitySpec(null, null, true),
+        null,
         null
     );
     Assert.assertFalse(MSQ_COMPACTION_RUNNER.validateCompactionTask(compactionTask, INTERVAL_DATASCHEMAS).isValid());
   }
 
   @Test
-  public void testMSQEngineWithUnsupportedMetricsSpecIsInValid()
+  public void testMSQEngineWithUnsupportedMetricsSpecIsInvalid()
   {
     // Aggregators having different input and ouput column names are unsupported.
     final String inputColName = "added";
@@ -276,6 +313,7 @@ public void testMSQEngineWithUnsupportedMetricsSpecIsInValid()
         null,
         Collections.emptyMap(),
         new ClientCompactionTaskGranularitySpec(null, null, true),
+        null,
         new AggregatorFactory[]{new LongSumAggregatorFactory(outputColName, inputColName)}
     );
     CompactionConfigValidationResult validationResult = MSQ_COMPACTION_RUNNER.validateCompactionTask(
@@ -292,7 +330,7 @@ public void testMSQEngineWithUnsupportedMetricsSpecIsInValid()
   @Test
   public void testRunCompactionTasksWithEmptyTaskListFails() throws Exception
   {
-    CompactionTask compactionTask = createCompactionTask(null, null, Collections.emptyMap(), null, null);
+    CompactionTask compactionTask = createCompactionTask(null, null, Collections.emptyMap(), null, null, null);
     TaskStatus taskStatus = MSQ_COMPACTION_RUNNER.runCompactionTasks(compactionTask, Collections.emptyMap(), null);
     Assert.assertTrue(taskStatus.isFailure());
   }
@@ -307,6 +345,7 @@ public void testMSQControllerTaskSpecWithScanIsValid() throws JsonProcessingExce
         dimFilter,
         Collections.emptyMap(),
         null,
+        null,
         null
     );
 
@@ -384,6 +423,7 @@ public void testMSQControllerTaskSpecWithAggregatorsIsValid() throws JsonProcess
         dimFilter,
         Collections.emptyMap(),
         null,
+        null,
         null
     );
 
@@ -481,6 +521,7 @@ private CompactionTask createCompactionTask(
       @Nullable DimFilter dimFilter,
       Map<String, Object> contextParams,
       @Nullable ClientCompactionTaskGranularitySpec granularitySpec,
+      @Nullable List<DimensionSchema> dimensionSchemas,
       @Nullable AggregatorFactory[] metricsSpec
   )
   {
@@ -504,6 +545,7 @@ private CompactionTask createCompactionTask(
         ))
         .transformSpec(transformSpec)
         .granularitySpec(granularitySpec)
+        .dimensionsSpec(new DimensionsSpec(dimensionSchemas))
         .metricsSpec(metricsSpec)
         .compactionRunner(MSQ_COMPACTION_RUNNER)
         .context(context);