From 35390e6d559e05b761f9971da8ec0aba64b49af2 Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Mon, 29 Jul 2024 09:53:00 +0530 Subject: [PATCH 1/5] Remove deprecated maxRowsPerSegment from batch task tuning configs --- .../k8s/overlord/common/K8sTestUtils.java | 2 - .../msq/indexing/MSQCompactionRunnerTest.java | 1 - .../indexing/common/task/CompactionTask.java | 24 --- .../druid/indexing/common/task/IndexTask.java | 164 ++++-------------- .../parallel/ParallelIndexSupervisorTask.java | 7 - .../parallel/ParallelIndexTuningConfig.java | 15 -- .../SeekableStreamIndexTaskTuningConfig.java | 1 - .../common/task/CompactionTaskRunTest.java | 6 +- .../common/task/CompactionTaskTest.java | 67 +------ .../common/task/IndexTaskSerdeTest.java | 93 ++-------- .../indexing/common/task/IndexTaskTest.java | 45 ++--- .../indexing/common/task/TaskSerdeTest.java | 46 ++--- .../common/task/TuningConfigBuilder.java | 50 ------ .../parallel/HashPartitionTaskKillTest.java | 9 +- .../indexing/overlord/TaskLifecycleTest.java | 11 +- .../appenderator/AppenderatorConfig.java | 9 - .../UnifiedIndexerAppenderatorsManager.java | 11 -- ...nifiedIndexerAppenderatorsManagerTest.java | 2 - .../cli/validate/DruidJsonValidatorTest.java | 7 - 19 files changed, 119 insertions(+), 451 deletions(-) diff --git a/extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/K8sTestUtils.java b/extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/K8sTestUtils.java index b3fda99b222e..3c91b8873e73 100644 --- a/extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/K8sTestUtils.java +++ b/extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/K8sTestUtils.java @@ -85,8 +85,6 @@ public static Task getTask() false ), new IndexTask.IndexTuningConfig( - null, - null, null, 10, null, diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java index b95243f7783f..f65315a79e39 100644 --- a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java +++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java @@ -402,7 +402,6 @@ private static CompactionTask.CompactionTuningConfig createTuningConfig( .forCompactionTask() .withMaxRowsInMemory(500000) .withMaxBytesInMemory(1000000L) - .withMaxTotalRows(Long.MAX_VALUE) .withPartitionsSpec(partitionsSpec) .withIndexSpec(indexSpec) .withForceGuaranteedRollup(!(partitionsSpec instanceof DynamicPartitionsSpec)) diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java index 8659eb0f397e..ae87514e1f2b 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java @@ -252,14 +252,10 @@ static CompactionTuningConfig getTuningConfig(TuningConfig tuningConfig) } else if (tuningConfig instanceof ParallelIndexTuningConfig) { final ParallelIndexTuningConfig parallelIndexTuningConfig = (ParallelIndexTuningConfig) tuningConfig; return new CompactionTuningConfig( - null, - parallelIndexTuningConfig.getMaxRowsPerSegment(), parallelIndexTuningConfig.getAppendableIndexSpec(), parallelIndexTuningConfig.getMaxRowsInMemory(), parallelIndexTuningConfig.getMaxBytesInMemory(), parallelIndexTuningConfig.isSkipBytesInMemoryOverheadCheck(), - parallelIndexTuningConfig.getMaxTotalRows(), - parallelIndexTuningConfig.getNumShards(), parallelIndexTuningConfig.getSplitHintSpec(), parallelIndexTuningConfig.getPartitionsSpec(), parallelIndexTuningConfig.getIndexSpec(), @@ -287,14 +283,10 @@ static CompactionTuningConfig getTuningConfig(TuningConfig tuningConfig) } else if (tuningConfig instanceof IndexTuningConfig) { final IndexTuningConfig indexTuningConfig = (IndexTuningConfig) tuningConfig; return new CompactionTuningConfig( - null, - indexTuningConfig.getMaxRowsPerSegment(), indexTuningConfig.getAppendableIndexSpec(), indexTuningConfig.getMaxRowsInMemory(), indexTuningConfig.getMaxBytesInMemory(), indexTuningConfig.isSkipBytesInMemoryOverheadCheck(), - indexTuningConfig.getMaxTotalRows(), - indexTuningConfig.getNumShards(), null, indexTuningConfig.getPartitionsSpec(), indexTuningConfig.getIndexSpec(), @@ -1184,10 +1176,6 @@ public static CompactionTuningConfig defaultConfig() null, null, null, - null, - null, - null, - null, 0L, null ); @@ -1195,14 +1183,10 @@ public static CompactionTuningConfig defaultConfig() @JsonCreator public CompactionTuningConfig( - @JsonProperty("targetPartitionSize") @Deprecated @Nullable Integer targetPartitionSize, - @JsonProperty("maxRowsPerSegment") @Deprecated @Nullable Integer maxRowsPerSegment, @JsonProperty("appendableIndexSpec") @Nullable AppendableIndexSpec appendableIndexSpec, @JsonProperty("maxRowsInMemory") @Nullable Integer maxRowsInMemory, @JsonProperty("maxBytesInMemory") @Nullable Long maxBytesInMemory, @JsonProperty("skipBytesInMemoryOverheadCheck") @Nullable Boolean skipBytesInMemoryOverheadCheck, - @JsonProperty("maxTotalRows") @Deprecated @Nullable Long maxTotalRows, - @JsonProperty("numShards") @Deprecated @Nullable Integer numShards, @JsonProperty("splitHintSpec") @Nullable SplitHintSpec splitHintSpec, @JsonProperty("partitionsSpec") @Nullable PartitionsSpec partitionsSpec, @JsonProperty("indexSpec") @Nullable IndexSpec indexSpec, @@ -1230,14 +1214,10 @@ public CompactionTuningConfig( ) { super( - targetPartitionSize, - maxRowsPerSegment, appendableIndexSpec, maxRowsInMemory, maxBytesInMemory, skipBytesInMemoryOverheadCheck, - maxTotalRows, - numShards, splitHintSpec, partitionsSpec, indexSpec, @@ -1276,14 +1256,10 @@ public CompactionTuningConfig( public CompactionTuningConfig withPartitionsSpec(PartitionsSpec partitionsSpec) { return new CompactionTuningConfig( - null, - null, getAppendableIndexSpec(), getMaxRowsInMemory(), getMaxBytesInMemory(), isSkipBytesInMemoryOverheadCheck(), - null, - null, getSplitHintSpec(), partitionsSpec, getIndexSpec(), diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java index a8f60cca5729..16744761e6ec 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java @@ -32,6 +32,7 @@ import com.google.common.hash.HashFunction; import com.google.common.hash.Hashing; import com.google.common.util.concurrent.ListenableFuture; +import org.apache.druid.common.config.Configs; import org.apache.druid.data.input.InputFormat; import org.apache.druid.data.input.InputRow; import org.apache.druid.data.input.InputSource; @@ -1120,8 +1121,8 @@ public static class IndexIOConfig implements BatchIOConfig private final InputSource inputSource; private final AtomicReference inputSourceWithToolbox = new AtomicReference<>(); private final InputFormat inputFormat; - private boolean appendToExisting; - private boolean dropExisting; + private final boolean appendToExisting; + private final boolean dropExisting; @JsonCreator public IndexIOConfig( @@ -1133,8 +1134,8 @@ public IndexIOConfig( { this.inputSource = inputSource; this.inputFormat = inputFormat; - this.appendToExisting = appendToExisting == null ? BatchIOConfig.DEFAULT_APPEND_EXISTING : appendToExisting; - this.dropExisting = dropExisting == null ? BatchIOConfig.DEFAULT_DROP_EXISTING : dropExisting; + this.appendToExisting = Configs.valueOrDefault(appendToExisting, BatchIOConfig.DEFAULT_APPEND_EXISTING); + this.dropExisting = Configs.valueOrDefault(dropExisting, BatchIOConfig.DEFAULT_DROP_EXISTING); } @Nullable @@ -1234,33 +1235,12 @@ public static class IndexTuningConfig implements AppenderatorConfig private final int numPersistThreads; - @Nullable - private static PartitionsSpec getPartitionsSpec( + private static void validatePartitionsSpec( boolean forceGuaranteedRollup, - @Nullable PartitionsSpec partitionsSpec, - @Nullable Integer maxRowsPerSegment, - @Nullable Long maxTotalRows, - @Nullable Integer numShards, - @Nullable List partitionDimensions + PartitionsSpec partitionsSpec ) { - if (partitionsSpec == null) { - if (forceGuaranteedRollup) { - if (maxRowsPerSegment != null - || numShards != null - || (partitionDimensions != null && !partitionDimensions.isEmpty())) { - return new HashedPartitionsSpec(maxRowsPerSegment, numShards, partitionDimensions); - } else { - return null; - } - } else { - if (maxRowsPerSegment != null || maxTotalRows != null) { - return new DynamicPartitionsSpec(maxRowsPerSegment, maxTotalRows); - } else { - return null; - } - } - } else { + if (partitionsSpec != null) { if (forceGuaranteedRollup) { if (!partitionsSpec.isForceGuaranteedRollupCompatibleType()) { throw new IAE(partitionsSpec.getClass().getSimpleName() + " cannot be used for perfect rollup"); @@ -1270,29 +1250,21 @@ private static PartitionsSpec getPartitionsSpec( throw new IAE("DynamicPartitionsSpec must be used for best-effort rollup"); } } - return partitionsSpec; } } @JsonCreator public IndexTuningConfig( - @JsonProperty("targetPartitionSize") @Deprecated @Nullable Integer targetPartitionSize, - @JsonProperty("maxRowsPerSegment") @Deprecated @Nullable Integer maxRowsPerSegment, @JsonProperty("appendableIndexSpec") @Nullable AppendableIndexSpec appendableIndexSpec, @JsonProperty("maxRowsInMemory") @Nullable Integer maxRowsInMemory, @JsonProperty("maxBytesInMemory") @Nullable Long maxBytesInMemory, @JsonProperty("skipBytesInMemoryOverheadCheck") @Nullable Boolean skipBytesInMemoryOverheadCheck, - @JsonProperty("maxTotalRows") @Deprecated @Nullable Long maxTotalRows, - @JsonProperty("rowFlushBoundary") @Deprecated @Nullable Integer rowFlushBoundary_forBackCompatibility, - @JsonProperty("numShards") @Deprecated @Nullable Integer numShards, - @JsonProperty("partitionDimensions") @Deprecated @Nullable List partitionDimensions, @JsonProperty("partitionsSpec") @Nullable PartitionsSpec partitionsSpec, @JsonProperty("indexSpec") @Nullable IndexSpec indexSpec, @JsonProperty("indexSpecForIntermediatePersists") @Nullable IndexSpec indexSpecForIntermediatePersists, @JsonProperty("maxPendingPersists") @Nullable Integer maxPendingPersists, @JsonProperty("forceGuaranteedRollup") @Nullable Boolean forceGuaranteedRollup, @JsonProperty("reportParseExceptions") @Deprecated @Nullable Boolean reportParseExceptions, - @JsonProperty("publishTimeout") @Deprecated @Nullable Long publishTimeout, @JsonProperty("pushTimeout") @Nullable Long pushTimeout, @JsonProperty("segmentWriteOutMediumFactory") @Nullable SegmentWriteOutMediumFactory segmentWriteOutMediumFactory, @@ -1306,23 +1278,16 @@ public IndexTuningConfig( { this( appendableIndexSpec, - maxRowsInMemory != null ? maxRowsInMemory : rowFlushBoundary_forBackCompatibility, - maxBytesInMemory != null ? maxBytesInMemory : 0, - skipBytesInMemoryOverheadCheck != null ? skipBytesInMemoryOverheadCheck : DEFAULT_SKIP_BYTES_IN_MEMORY_OVERHEAD_CHECK, - getPartitionsSpec( - forceGuaranteedRollup == null ? DEFAULT_GUARANTEE_ROLLUP : forceGuaranteedRollup, - partitionsSpec, - maxRowsPerSegment == null ? targetPartitionSize : maxRowsPerSegment, - maxTotalRows, - numShards, - partitionDimensions - ), + maxRowsInMemory, + maxBytesInMemory, + skipBytesInMemoryOverheadCheck, + partitionsSpec, indexSpec, indexSpecForIntermediatePersists, maxPendingPersists, forceGuaranteedRollup, reportParseExceptions, - pushTimeout != null ? pushTimeout : publishTimeout, + pushTimeout, null, segmentWriteOutMediumFactory, logParseExceptions, @@ -1332,11 +1297,6 @@ public IndexTuningConfig( awaitSegmentAvailabilityTimeoutMillis, numPersistThreads ); - - Preconditions.checkArgument( - targetPartitionSize == null || maxRowsPerSegment == null, - "Can't use targetPartitionSize and maxRowsPerSegment together" - ); } private IndexTuningConfig() @@ -1366,26 +1326,23 @@ private IndexTuningConfig( @Nullable Integer numPersistThreads ) { - this.appendableIndexSpec = appendableIndexSpec == null ? DEFAULT_APPENDABLE_INDEX : appendableIndexSpec; - this.maxRowsInMemory = maxRowsInMemory == null ? TuningConfig.DEFAULT_MAX_ROWS_IN_MEMORY_BATCH : maxRowsInMemory; + this.appendableIndexSpec = Configs.valueOrDefault(appendableIndexSpec, DEFAULT_APPENDABLE_INDEX); + this.maxRowsInMemory = Configs.valueOrDefault(maxRowsInMemory, TuningConfig.DEFAULT_MAX_ROWS_IN_MEMORY_BATCH); // initializing this to 0, it will be lazily initialized to a value // @see #getMaxBytesInMemoryOrDefault() - this.maxBytesInMemory = maxBytesInMemory == null ? 0 : maxBytesInMemory; - this.skipBytesInMemoryOverheadCheck = skipBytesInMemoryOverheadCheck == null ? - DEFAULT_SKIP_BYTES_IN_MEMORY_OVERHEAD_CHECK : skipBytesInMemoryOverheadCheck; - this.maxColumnsToMerge = maxColumnsToMerge == null - ? IndexMerger.UNLIMITED_MAX_COLUMNS_TO_MERGE - : maxColumnsToMerge; + this.maxBytesInMemory = Configs.valueOrDefault(maxBytesInMemory, 0); + this.skipBytesInMemoryOverheadCheck = Configs.valueOrDefault( + skipBytesInMemoryOverheadCheck, + DEFAULT_SKIP_BYTES_IN_MEMORY_OVERHEAD_CHECK + ); + this.maxColumnsToMerge = Configs.valueOrDefault(maxColumnsToMerge, IndexMerger.UNLIMITED_MAX_COLUMNS_TO_MERGE); this.partitionsSpec = partitionsSpec; - this.indexSpec = indexSpec == null ? DEFAULT_INDEX_SPEC : indexSpec; - this.indexSpecForIntermediatePersists = indexSpecForIntermediatePersists == null ? - this.indexSpec : indexSpecForIntermediatePersists; - this.maxPendingPersists = maxPendingPersists == null ? DEFAULT_MAX_PENDING_PERSISTS : maxPendingPersists; - this.forceGuaranteedRollup = forceGuaranteedRollup == null ? DEFAULT_GUARANTEE_ROLLUP : forceGuaranteedRollup; - this.reportParseExceptions = reportParseExceptions == null - ? DEFAULT_REPORT_PARSE_EXCEPTIONS - : reportParseExceptions; - this.pushTimeout = pushTimeout == null ? DEFAULT_PUSH_TIMEOUT : pushTimeout; + this.indexSpec = Configs.valueOrDefault(indexSpec, DEFAULT_INDEX_SPEC); + this.indexSpecForIntermediatePersists = Configs.valueOrDefault(indexSpecForIntermediatePersists, this.indexSpec); + this.maxPendingPersists = Configs.valueOrDefault(maxPendingPersists, DEFAULT_MAX_PENDING_PERSISTS); + this.forceGuaranteedRollup = Configs.valueOrDefault(forceGuaranteedRollup, DEFAULT_GUARANTEE_ROLLUP); + this.reportParseExceptions = Configs.valueOrDefault(reportParseExceptions, DEFAULT_REPORT_PARSE_EXCEPTIONS); + this.pushTimeout = Configs.valueOrDefault(pushTimeout, DEFAULT_PUSH_TIMEOUT); this.basePersistDirectory = basePersistDirectory; this.segmentWriteOutMediumFactory = segmentWriteOutMediumFactory; @@ -1394,16 +1351,16 @@ private IndexTuningConfig( this.maxParseExceptions = 0; this.maxSavedParseExceptions = maxSavedParseExceptions == null ? 0 : Math.min(1, maxSavedParseExceptions); } else { - this.maxParseExceptions = maxParseExceptions == null - ? TuningConfig.DEFAULT_MAX_PARSE_EXCEPTIONS - : maxParseExceptions; - this.maxSavedParseExceptions = maxSavedParseExceptions == null - ? TuningConfig.DEFAULT_MAX_SAVED_PARSE_EXCEPTIONS - : maxSavedParseExceptions; + this.maxParseExceptions = Configs.valueOrDefault( + maxParseExceptions, + TuningConfig.DEFAULT_MAX_PARSE_EXCEPTIONS + ); + this.maxSavedParseExceptions = Configs.valueOrDefault( + maxSavedParseExceptions, + TuningConfig.DEFAULT_MAX_SAVED_PARSE_EXCEPTIONS + ); } - this.logParseExceptions = logParseExceptions == null - ? TuningConfig.DEFAULT_LOG_PARSE_EXCEPTIONS - : logParseExceptions; + this.logParseExceptions = Configs.valueOrDefault(logParseExceptions, TuningConfig.DEFAULT_LOG_PARSE_EXCEPTIONS); if (awaitSegmentAvailabilityTimeoutMillis == null || awaitSegmentAvailabilityTimeoutMillis < 0) { this.awaitSegmentAvailabilityTimeoutMillis = DEFAULT_AWAIT_SEGMENT_AVAILABILITY_TIMEOUT_MILLIS; } else { @@ -1411,6 +1368,8 @@ private IndexTuningConfig( } this.numPersistThreads = numPersistThreads == null ? DEFAULT_NUM_PERSIST_THREADS : Math.max(numPersistThreads, DEFAULT_NUM_PERSIST_THREADS); + + validatePartitionsSpec(this.forceGuaranteedRollup, this.partitionsSpec); } @Override @@ -1559,53 +1518,6 @@ public int getMaxSavedParseExceptions() return maxSavedParseExceptions; } - /** - * Return the max number of rows per segment. This returns null if it's not specified in tuningConfig. - * Deprecated in favor of {@link #getGivenOrDefaultPartitionsSpec()}. - */ - @Nullable - @Override - @Deprecated - @JsonProperty - public Integer getMaxRowsPerSegment() - { - return partitionsSpec == null ? null : partitionsSpec.getMaxRowsPerSegment(); - } - - /** - * Return the max number of total rows in appenderator. This returns null if it's not specified in tuningConfig. - * Deprecated in favor of {@link #getGivenOrDefaultPartitionsSpec()}. - */ - @Override - @Nullable - @Deprecated - @JsonProperty - public Long getMaxTotalRows() - { - return partitionsSpec instanceof DynamicPartitionsSpec - ? ((DynamicPartitionsSpec) partitionsSpec).getMaxTotalRows() - : null; - } - - @Deprecated - @Nullable - @JsonProperty - public Integer getNumShards() - { - return partitionsSpec instanceof HashedPartitionsSpec - ? ((HashedPartitionsSpec) partitionsSpec).getNumShards() - : null; - } - - @Deprecated - @JsonProperty - public List getPartitionDimensions() - { - return partitionsSpec instanceof HashedPartitionsSpec - ? ((HashedPartitionsSpec) partitionsSpec).getPartitionDimensions() - : Collections.emptyList(); - } - @Override public File getBasePersistDirectory() { diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java index 6039a62bc984..a791aae0e942 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java @@ -1287,23 +1287,16 @@ private void writeCompletionReports() private static IndexTuningConfig convertToIndexTuningConfig(ParallelIndexTuningConfig tuningConfig) { return new IndexTuningConfig( - null, - null, tuningConfig.getAppendableIndexSpec(), tuningConfig.getMaxRowsInMemory(), tuningConfig.getMaxBytesInMemory(), tuningConfig.isSkipBytesInMemoryOverheadCheck(), - null, - null, - null, - null, tuningConfig.getPartitionsSpec(), tuningConfig.getIndexSpec(), tuningConfig.getIndexSpecForIntermediatePersists(), tuningConfig.getMaxPendingPersists(), tuningConfig.isForceGuaranteedRollup(), tuningConfig.isReportParseExceptions(), - null, tuningConfig.getPushTimeout(), tuningConfig.getSegmentWriteOutMediumFactory(), tuningConfig.isLogParseExceptions(), diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexTuningConfig.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexTuningConfig.java index d97d9beff343..4ff3a3994e27 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexTuningConfig.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexTuningConfig.java @@ -104,24 +104,16 @@ public static ParallelIndexTuningConfig defaultConfig() null, null, null, - null, - null, - null, - null, null ); } @JsonCreator public ParallelIndexTuningConfig( - @JsonProperty("targetPartitionSize") @Deprecated @Nullable Integer targetPartitionSize, - @JsonProperty("maxRowsPerSegment") @Deprecated @Nullable Integer maxRowsPerSegment, @JsonProperty("appendableIndexSpec") @Nullable AppendableIndexSpec appendableIndexSpec, @JsonProperty("maxRowsInMemory") @Nullable Integer maxRowsInMemory, @JsonProperty("maxBytesInMemory") @Nullable Long maxBytesInMemory, @JsonProperty("skipBytesInMemoryOverheadCheck") @Nullable Boolean skipBytesInMemoryOverheadCheck, - @JsonProperty("maxTotalRows") @Deprecated @Nullable Long maxTotalRows, - @JsonProperty("numShards") @Deprecated @Nullable Integer numShards, @JsonProperty("splitHintSpec") @Nullable SplitHintSpec splitHintSpec, @JsonProperty("partitionsSpec") @Nullable PartitionsSpec partitionsSpec, @JsonProperty("indexSpec") @Nullable IndexSpec indexSpec, @@ -149,23 +141,16 @@ public ParallelIndexTuningConfig( ) { super( - targetPartitionSize, - maxRowsPerSegment, appendableIndexSpec, maxRowsInMemory, maxBytesInMemory, skipBytesInMemoryOverheadCheck, - maxTotalRows, - null, - numShards, - null, partitionsSpec, indexSpec, indexSpecForIntermediatePersists, maxPendingPersists, forceGuaranteedRollup, reportParseExceptions, - null, pushTimeout, segmentWriteOutMediumFactory, logParseExceptions, diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTuningConfig.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTuningConfig.java index 34e4c5f0f2da..ce3254614173 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTuningConfig.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskTuningConfig.java @@ -169,7 +169,6 @@ public boolean isSkipBytesInMemoryOverheadCheck() return skipBytesInMemoryOverheadCheck; } - @Override @JsonProperty public Integer getMaxRowsPerSegment() { diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java index 7ebde700bee2..20ba9c263fe9 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskRunTest.java @@ -545,7 +545,7 @@ public void testRunIndexAndCompactAtTheSameTimeForDifferentInterval() throws Exc Granularities.MINUTE, null ), - IndexTaskTest.createTuningConfig(2, 2, 2L, null, false, true), + IndexTaskTest.createTuningConfig(2, 2, 2L, false, true), false, false ), @@ -1918,7 +1918,7 @@ private Pair runIndexTask( Granularities.MINUTE, null ), - IndexTaskTest.createTuningConfig(2, 2, 2L, null, false, true), + IndexTaskTest.createTuningConfig(2, 2, 2L, false, true), appendToExisting, false ), @@ -1958,7 +1958,7 @@ private Pair runIndexTask( Granularities.MINUTE, null ), - IndexTaskTest.createTuningConfig(2, 2, 2L, null, false, true), + IndexTaskTest.createTuningConfig(2, 2, 2L, false, true), appendToExisting, false ), diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java index f9849b1483df..99e66af9a307 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskTest.java @@ -55,6 +55,7 @@ import org.apache.druid.guice.GuiceInjectors; import org.apache.druid.guice.IndexingServiceTuningConfigModule; import org.apache.druid.indexer.TaskStatus; +import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; import org.apache.druid.indexer.partitions.HashedPartitionsSpec; import org.apache.druid.indexing.common.LockGranularity; import org.apache.druid.indexing.common.SegmentCacheManagerFactory; @@ -586,7 +587,6 @@ public void testSerdeWithOldTuningConfigSuccessfullyDeserializeToNewOne() throws ) .withForceGuaranteedRollup(true) .withReportParseExceptions(false) - .withPublishTimeout(5000L) .build(), null, toolbox.getJsonMapper(), @@ -781,68 +781,6 @@ public void testCreateIngestionSchema() throws IOException ); } - @Test - public void testCreateIngestionSchemaWithTargetPartitionSize() throws IOException - { - final CompactionTask.CompactionTuningConfig tuningConfig = TuningConfigBuilder - .forCompactionTask() - .withTargetPartitionSize(100000) - .withMaxRowsInMemory(500000) - .withMaxBytesInMemory(1000000L) - .withIndexSpec( - IndexSpec.builder() - .withBitmapSerdeFactory(RoaringBitmapSerdeFactory.getInstance()) - .withDimensionCompression(CompressionStrategy.LZ4) - .withMetricCompression(CompressionStrategy.LZF) - .withLongEncoding(LongEncodingStrategy.LONGS) - .build() - ) - .withForceGuaranteedRollup(true) - .withReportParseExceptions(false) - .withMaxNumConcurrentSubTasks(10) - .build(); - - final Map dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals( - toolbox, - LockGranularity.TIME_CHUNK, - new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)), - null, - null, - null, - null, - METRIC_BUILDER - ); - - final List ingestionSpecs = NativeCompactionRunner.createIngestionSpecs( - dataSchemasForIntervals, - toolbox, - new CompactionIOConfig(null, false, null), - new PartitionConfigurationManager(tuningConfig), - COORDINATOR_CLIENT, - segmentCacheManagerFactory - ); - - final List expectedDimensionsSpec = getExpectedDimensionsSpecForAutoGeneration(); - - ingestionSpecs.sort( - (s1, s2) -> Comparators.intervalsByStartThenEnd().compare( - s1.getDataSchema().getGranularitySpec().inputIntervals().get(0), - s2.getDataSchema().getGranularitySpec().inputIntervals().get(0) - ) - ); - Assert.assertEquals(6, ingestionSpecs.size()); - assertIngestionSchema( - ingestionSpecs, - expectedDimensionsSpec, - AGGREGATORS.stream().map(AggregatorFactory::getCombiningFactory).collect(Collectors.toList()), - SEGMENT_INTERVALS, - tuningConfig, - Granularities.MONTH, - Granularities.NONE, - BatchIOConfig.DEFAULT_DROP_EXISTING - ); - } - @Test public void testCreateIngestionSchemaWithMaxTotalRows() throws IOException { @@ -850,7 +788,7 @@ public void testCreateIngestionSchemaWithMaxTotalRows() throws IOException .forCompactionTask() .withMaxRowsInMemory(500000) .withMaxBytesInMemory(1000000L) - .withMaxTotalRows(1000000L) + .withPartitionsSpec(new DynamicPartitionsSpec(null, 1000000L)) .withIndexSpec( IndexSpec.builder() .withBitmapSerdeFactory(RoaringBitmapSerdeFactory.getInstance()) @@ -1641,7 +1579,6 @@ private void assertIngestionSchema( .forCompactionTask() .withMaxRowsInMemory(500000) .withMaxBytesInMemory(1000000L) - .withMaxTotalRows(Long.MAX_VALUE) .withPartitionsSpec(new HashedPartitionsSpec(5000000, null, null)) .withIndexSpec( IndexSpec.builder() diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskSerdeTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskSerdeTest.java index 493f6b477565..cfbb2695bf13 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskSerdeTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskSerdeTest.java @@ -21,22 +21,14 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.jsontype.NamedType; -import com.google.common.collect.ImmutableList; import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; import org.apache.druid.indexer.partitions.HashedPartitionsSpec; import org.apache.druid.indexing.common.task.IndexTask.IndexTuningConfig; import org.apache.druid.jackson.DefaultObjectMapper; -import org.apache.druid.segment.IndexSpec; -import org.apache.druid.segment.data.CompressionFactory.LongEncodingStrategy; -import org.apache.druid.segment.data.CompressionStrategy; -import org.apache.druid.segment.data.RoaringBitmapSerdeFactory; import org.apache.druid.segment.indexing.TuningConfig; -import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory; import org.junit.Assert; import org.junit.BeforeClass; -import org.junit.Rule; import org.junit.Test; -import org.junit.rules.ExpectedException; import java.io.IOException; @@ -44,9 +36,6 @@ public class IndexTaskSerdeTest { private static final ObjectMapper MAPPER = new DefaultObjectMapper(); - @Rule - public ExpectedException expectedException = ExpectedException.none(); - @BeforeClass public static void setup() { @@ -75,80 +64,30 @@ public void testSerdeTuningConfigWithHashedPartitionsSpec() throws IOException assertSerdeTuningConfig(tuningConfig); } - @Test - public void testSerdeTuningConfigWithDeprecatedDynamicPartitionsSpec() throws IOException - { - final IndexTuningConfig tuningConfig = TuningConfigBuilder - .forIndexTask() - .withMaxRowsPerSegment(1000) - .withMaxRowsInMemory(100) - .withMaxBytesInMemory(2000L) - .withMaxTotalRows(3000L) - .withForceGuaranteedRollup(false) - .withPushTimeout(100L) - .withAwaitSegmentAvailabilityTimeoutMillis(1L) - .build(); - assertSerdeTuningConfig(tuningConfig); - } - - @Test - public void testSerdeTuningConfigWithDeprecatedHashedPartitionsSpec() throws IOException - { - final IndexTuningConfig tuningConfig = new IndexTuningConfig( - null, - null, - null, - 100, - 2000L, - null, - null, - null, - 10, - ImmutableList.of("dim1", "dim2"), - null, - IndexSpec.builder() - .withBitmapSerdeFactory(RoaringBitmapSerdeFactory.getInstance()) - .withDimensionCompression(CompressionStrategy.LZ4) - .withMetricCompression(CompressionStrategy.LZF) - .withLongEncoding(LongEncodingStrategy.LONGS) - .build(), - null, - null, - false, - null, - null, - 100L, - OffHeapMemorySegmentWriteOutMediumFactory.instance(), - true, - 10, - 100, - 1234, - null, - null - ); - assertSerdeTuningConfig(tuningConfig); - } - @Test public void testForceGuaranteedRollupWithDynamicPartitionsSpec() { - expectedException.expect(IllegalArgumentException.class); - expectedException.expectMessage("DynamicPartitionsSpec cannot be used for perfect rollup"); - TuningConfigBuilder.forIndexTask() - .withForceGuaranteedRollup(true) - .withPartitionsSpec(new DynamicPartitionsSpec(1000, 2000L)) - .build(); + Exception e = Assert.assertThrows( + IllegalArgumentException.class, + () -> TuningConfigBuilder.forIndexTask() + .withForceGuaranteedRollup(true) + .withPartitionsSpec(new DynamicPartitionsSpec(1000, 2000L)) + .build() + ); + Assert.assertEquals("DynamicPartitionsSpec cannot be used for perfect rollup", e.getMessage()); } @Test public void testBestEffortRollupWithHashedPartitionsSpec() { - expectedException.expect(IllegalArgumentException.class); - expectedException.expectMessage("DynamicPartitionsSpec must be used for best-effort rollup"); - TuningConfigBuilder.forIndexTask() - .withForceGuaranteedRollup(false) - .withPartitionsSpec(new HashedPartitionsSpec(null, 10, null)) - .build(); + Exception e = Assert.assertThrows( + IllegalArgumentException.class, + () -> TuningConfigBuilder.forIndexTask() + .withForceGuaranteedRollup(false) + .withPartitionsSpec(new HashedPartitionsSpec(null, 10, null)) + .build() + ); + Assert.assertEquals("DynamicPartitionsSpec must be used for best-effort rollup", e.getMessage()); } private static void assertSerdeTuningConfig(IndexTuningConfig tuningConfig) throws IOException diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java index 3ec5b8d0aa8a..e2e5d3ceaeea 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/IndexTaskTest.java @@ -1041,7 +1041,7 @@ public void testWithSmallMaxTotalRows() throws Exception Granularities.MINUTE, null ), - createTuningConfig(2, 2, 2L, null, false, true), + createTuningConfig(2, 2, 2L, false, true), false, false ), @@ -1078,7 +1078,7 @@ public void testPerfectRollup() throws Exception true, null ), - createTuningConfig(3, 2, 2L, null, true, true), + createTuningConfig(3, 2, 2L, true, true), false, false ), @@ -1114,7 +1114,7 @@ public void testBestEffortRollup() throws Exception true, null ), - createTuningConfig(3, 2, 2L, null, false, true), + createTuningConfig(3, 2, 2L, false, true), false, false ), @@ -1375,7 +1375,7 @@ public void testIgnoreParseException() throws Exception final TimestampSpec timestampSpec = new TimestampSpec("time", "auto", null); final List columns = Arrays.asList("time", "dim", "val"); // ignore parse exception - final IndexTuningConfig tuningConfig = createTuningConfig(2, null, null, null, false, false); + final IndexTuningConfig tuningConfig = createTuningConfig(2, null, null, false, false); // GranularitySpec.intervals and numShards must be null to verify reportParseException=false is respected both in // IndexTask.determineShardSpecs() and IndexTask.generateAndPublishSegments() @@ -1427,7 +1427,7 @@ public void testReportParseException() throws Exception final TimestampSpec timestampSpec = new TimestampSpec("time", "auto", null); final List columns = Arrays.asList("time", "dim", "val"); // report parse exception - final IndexTuningConfig tuningConfig = createTuningConfig(2, null, null, null, false, true); + final IndexTuningConfig tuningConfig = createTuningConfig(2, null, null, false, true); final IndexIngestionSpec indexIngestionSpec; List expectedMessages; if (useInputFormatApi) { @@ -1882,7 +1882,7 @@ public void testCsvWithHeaderOfEmptyColumns() throws Exception } // report parse exception - final IndexTuningConfig tuningConfig = createTuningConfig(2, 1, null, null, true, true); + final IndexTuningConfig tuningConfig = createTuningConfig(2, 1, null, true, true); final IndexIngestionSpec ingestionSpec; if (useInputFormatApi) { ingestionSpec = createIngestionSpec( @@ -1951,7 +1951,7 @@ public void testCsvWithHeaderOfEmptyTimestamp() throws Exception final List columns = Arrays.asList("ts", "", ""); // report parse exception - final IndexTuningConfig tuningConfig = createTuningConfig(2, null, null, null, false, true); + final IndexTuningConfig tuningConfig = createTuningConfig(2, null, null, false, true); final IndexIngestionSpec ingestionSpec; List expectedMessages; if (useInputFormatApi) { @@ -2020,7 +2020,7 @@ public void testOverwriteWithSameSegmentGranularity() throws Exception true, null ), - createTuningConfig(3, 2, 2L, null, false, true), + createTuningConfig(3, 2, 2L, false, true), false, false ), @@ -2078,7 +2078,7 @@ public void testOverwriteWithDifferentSegmentGranularity() throws Exception true, null ), - createTuningConfig(3, 2, 2L, null, false, true), + createTuningConfig(3, 2, 2L, false, true), false, false ), @@ -2528,7 +2528,6 @@ private static IndexTuningConfig createTuningConfigWithMaxRowsPerSegment( maxRowsPerSegment, 1, null, - null, forceGuaranteedRollup, true ); @@ -2538,30 +2537,32 @@ private static IndexTuningConfig createTuningConfigWithPartitionsSpec( PartitionsSpec partitionsSpec ) { - return createTuningConfig( - null, - 1, - null, - partitionsSpec, - true, - true - ); + return TuningConfigBuilder + .forIndexTask() + .withMaxRowsInMemory(1) + .withPartitionsSpec(partitionsSpec) + .withIndexSpec(INDEX_SPEC) + .withForceGuaranteedRollup(true) + .withReportParseExceptions(true) + .withMaxSavedParseExceptions(1) + .build(); } static IndexTuningConfig createTuningConfig( - @Nullable Integer maxRowsPerSegment, + int maxRowsPerSegment, @Nullable Integer maxRowsInMemory, @Nullable Long maxTotalRows, - @Nullable PartitionsSpec partitionsSpec, boolean forceGuaranteedRollup, boolean reportParseException ) { + final PartitionsSpec partitionsSpec = forceGuaranteedRollup + ? new HashedPartitionsSpec(maxRowsPerSegment, null, null) + : new DynamicPartitionsSpec(maxRowsPerSegment, maxTotalRows); return TuningConfigBuilder .forIndexTask() - .withMaxRowsPerSegment(maxRowsPerSegment) .withMaxRowsInMemory(maxRowsInMemory) - .withMaxTotalRows(maxTotalRows) + //.withMaxTotalRows(maxTotalRows) .withPartitionsSpec(partitionsSpec) .withIndexSpec(INDEX_SPEC) .withForceGuaranteedRollup(forceGuaranteedRollup) diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskSerdeTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskSerdeTest.java index e6ea0e1329ae..e5a03a9c7f53 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskSerdeTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskSerdeTest.java @@ -79,8 +79,8 @@ public void testIndexTaskIOConfigDefaults() throws Exception IndexTask.IndexIOConfig.class ); - Assert.assertEquals(false, ioConfig.isAppendToExisting()); - Assert.assertEquals(false, ioConfig.isDropExisting()); + Assert.assertFalse(ioConfig.isAppendToExisting()); + Assert.assertFalse(ioConfig.isDropExisting()); } @Test @@ -96,8 +96,8 @@ public void testIndexTaskTuningConfigDefaults() throws Exception Assert.assertEquals(new Period(Integer.MAX_VALUE), tuningConfig.getIntermediatePersistPeriod()); Assert.assertEquals(0, tuningConfig.getMaxPendingPersists()); Assert.assertEquals(1000000, tuningConfig.getMaxRowsInMemory()); - Assert.assertNull(tuningConfig.getNumShards()); - Assert.assertNull(tuningConfig.getMaxRowsPerSegment()); + // TODO: Assert.assertNull(tuningConfig.getNumShards()); + Assert.assertNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); } @Test @@ -108,56 +108,59 @@ public void testIndexTaskTuningConfigTargetPartitionSizeOrNumShards() throws Exc IndexTask.IndexTuningConfig.class ); - Assert.assertEquals(10, (int) tuningConfig.getMaxRowsPerSegment()); - Assert.assertNull(tuningConfig.getNumShards()); + Assert.assertEquals(10, (int) tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); + // Assert.assertNull(tuningConfig.getNumShards()); tuningConfig = jsonMapper.readValue( "{\"type\":\"index\"}", IndexTask.IndexTuningConfig.class ); - Assert.assertNull(tuningConfig.getMaxRowsPerSegment()); + Assert.assertNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); tuningConfig = jsonMapper.readValue( "{\"type\":\"index\", \"maxRowsPerSegment\":10}", IndexTask.IndexTuningConfig.class ); - Assert.assertEquals(10, (int) tuningConfig.getMaxRowsPerSegment()); - Assert.assertNull(tuningConfig.getNumShards()); + Assert.assertEquals(10, (int) tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); + // Assert.assertNull(tuningConfig.getNumShards()); tuningConfig = jsonMapper.readValue( "{\"type\":\"index\", \"numShards\":10, \"forceGuaranteedRollup\": true}", IndexTask.IndexTuningConfig.class ); - Assert.assertNull(tuningConfig.getMaxRowsPerSegment()); - Assert.assertEquals(10, (int) tuningConfig.getNumShards()); + Assert.assertNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); + // Assert.assertEquals(10, (int) tuningConfig.getNumShards()); tuningConfig = jsonMapper.readValue( "{\"type\":\"index\", \"targetPartitionSize\":-1, \"numShards\":10, \"forceGuaranteedRollup\": true}", IndexTask.IndexTuningConfig.class ); - Assert.assertNull(tuningConfig.getMaxRowsPerSegment()); - Assert.assertEquals(10, (int) tuningConfig.getNumShards()); + Assert.assertNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); + // Assert.assertEquals(10, (int) tuningConfig.getNumShards()); tuningConfig = jsonMapper.readValue( "{\"type\":\"index\", \"targetPartitionSize\":10, \"numShards\":-1}", IndexTask.IndexTuningConfig.class ); - Assert.assertNull(tuningConfig.getNumShards()); - Assert.assertEquals(10, (int) tuningConfig.getMaxRowsPerSegment()); + // Assert.assertNull(tuningConfig.getNumShards()); + Assert.assertEquals(10, (int) tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); tuningConfig = jsonMapper.readValue( "{\"type\":\"index\", \"targetPartitionSize\":-1, \"numShards\":-1, \"forceGuaranteedRollup\": true}", IndexTask.IndexTuningConfig.class ); - Assert.assertNull(tuningConfig.getNumShards()); - Assert.assertNotNull(tuningConfig.getMaxRowsPerSegment()); - Assert.assertEquals(PartitionsSpec.DEFAULT_MAX_ROWS_PER_SEGMENT, tuningConfig.getMaxRowsPerSegment().intValue()); + // Assert.assertNull(tuningConfig.getNumShards()); + Assert.assertNotNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); + Assert.assertEquals( + PartitionsSpec.DEFAULT_MAX_ROWS_PER_SEGMENT, + tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment().intValue() + ); } @Test @@ -275,8 +278,11 @@ public void testIndexTaskSerde() throws Exception ); Assert.assertEquals(taskTuningConfig.getMaxPendingPersists(), task2TuningConfig.getMaxPendingPersists()); Assert.assertEquals(taskTuningConfig.getMaxRowsInMemory(), task2TuningConfig.getMaxRowsInMemory()); - Assert.assertEquals(taskTuningConfig.getNumShards(), task2TuningConfig.getNumShards()); - Assert.assertEquals(taskTuningConfig.getMaxRowsPerSegment(), task2TuningConfig.getMaxRowsPerSegment()); + // Assert.assertEquals(taskTuningConfig.getNumShards(), task2TuningConfig.getNumShards()); + Assert.assertEquals( + taskTuningConfig.getGivenOrDefaultPartitionsSpec(), + task2TuningConfig.getGivenOrDefaultPartitionsSpec() + ); Assert.assertEquals(taskTuningConfig.isReportParseExceptions(), task2TuningConfig.isReportParseExceptions()); Assert.assertEquals(taskTuningConfig.getAwaitSegmentAvailabilityTimeoutMillis(), task2TuningConfig.getAwaitSegmentAvailabilityTimeoutMillis()); } diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TuningConfigBuilder.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TuningConfigBuilder.java index c46c0becc76a..2aac155a0a2f 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TuningConfigBuilder.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TuningConfigBuilder.java @@ -37,14 +37,10 @@ */ public abstract class TuningConfigBuilder { - protected Integer targetPartitionSize; - protected Integer maxRowsPerSegment; protected AppendableIndexSpec appendableIndexSpec; protected Integer maxRowsInMemory; protected Long maxBytesInMemory; protected Boolean skipBytesInMemoryOverheadCheck; - protected Long maxTotalRows; - protected Integer numShards; protected SplitHintSpec splitHintSpec; protected PartitionsSpec partitionsSpec; protected IndexSpec indexSpec; @@ -52,7 +48,6 @@ public abstract class TuningConfigBuilder protected Integer maxPendingPersists; protected Boolean forceGuaranteedRollup; protected Boolean reportParseExceptions; - protected Long publishTimeout; protected Long pushTimeout; protected SegmentWriteOutMediumFactory segmentWriteOutMediumFactory; protected Integer maxNumSubTasks; @@ -71,18 +66,6 @@ public abstract class TuningConfigBuilder protected Integer maxAllowedLockCount; protected Integer numPersistThreads; - public TuningConfigBuilder withTargetPartitionSize(Integer targetPartitionSize) - { - this.targetPartitionSize = targetPartitionSize; - return this; - } - - public TuningConfigBuilder withMaxRowsPerSegment(Integer maxRowsPerSegment) - { - this.maxRowsPerSegment = maxRowsPerSegment; - return this; - } - public TuningConfigBuilder withAppendableIndexSpec(AppendableIndexSpec appendableIndexSpec) { this.appendableIndexSpec = appendableIndexSpec; @@ -107,18 +90,6 @@ public TuningConfigBuilder withSkipBytesInMemoryOverheadCheck(Boolean skipByt return this; } - public TuningConfigBuilder withMaxTotalRows(Long maxTotalRows) - { - this.maxTotalRows = maxTotalRows; - return this; - } - - public TuningConfigBuilder withNumShards(Integer numShards) - { - this.numShards = numShards; - return this; - } - public TuningConfigBuilder withSplitHintSpec(SplitHintSpec splitHintSpec) { this.splitHintSpec = splitHintSpec; @@ -167,12 +138,6 @@ public TuningConfigBuilder withPushTimeout(Long pushTimeout) return this; } - public TuningConfigBuilder withPublishTimeout(Long publishTimeout) - { - this.publishTimeout = publishTimeout; - return this; - } - public TuningConfigBuilder withSegmentWriteOutMediumFactory(SegmentWriteOutMediumFactory segmentWriteOutMediumFactory) { this.segmentWriteOutMediumFactory = segmentWriteOutMediumFactory; @@ -301,23 +266,16 @@ public static class Index extends TuningConfigBuilder ParallelIndexTaskRunner createRu // will break requiring messing with the logic below). // For the other two subsequent failures we need to have numShards non-null, so it bypasses // the first failure, so the conditions for failure in the different phase are given below: + final HashedPartitionsSpec partitionsSpec + = (HashedPartitionsSpec) getIngestionSchema().getTuningConfig().getPartitionsSpec(); + ParallelIndexTaskRunner retVal; if (succeedsBeforeFailing == 0 - && this.getIngestionSchema().getTuningConfig().getNumShards() == null) { + && partitionsSpec == null || partitionsSpec.getNumShards() == null) { retVal = (ParallelIndexTaskRunner) new TestRunner(false, "PHASE-1"); } else if (succeedsBeforeFailing == 0 - && this.getIngestionSchema().getTuningConfig().getNumShards() != null) { + && partitionsSpec.getNumShards() != null) { retVal = (ParallelIndexTaskRunner) new TestRunner(false, "PHASE-2"); } else if (succeedsBeforeFailing == 1 && numRuns == 1 - && this.getIngestionSchema().getTuningConfig().getNumShards() != null) { + && partitionsSpec == null || partitionsSpec.getNumShards() != null) { retVal = (ParallelIndexTaskRunner) new TestRunner(false, "PHASE-3"); } else { numRuns++; diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java index ddbed6be7c74..7df3292fbe8b 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java @@ -52,6 +52,7 @@ import org.apache.druid.indexer.TaskLocation; import org.apache.druid.indexer.TaskState; import org.apache.druid.indexer.TaskStatus; +import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; import org.apache.druid.indexing.common.SegmentCacheManagerFactory; import org.apache.druid.indexing.common.TaskLock; import org.apache.druid.indexing.common.TaskLockType; @@ -120,7 +121,6 @@ import org.apache.druid.segment.handoff.SegmentHandoffNotifierFactory; import org.apache.druid.segment.indexing.DataSchema; import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec; -import org.apache.druid.segment.join.JoinableFactoryWrapperTest; import org.apache.druid.segment.join.NoopJoinableFactory; import org.apache.druid.segment.loading.DataSegmentPusher; import org.apache.druid.segment.loading.LocalDataSegmentKiller; @@ -686,7 +686,7 @@ public void testIndexTask() ), new IndexIOConfig(new MockInputSource(), new NoopInputFormat(), false, false), TuningConfigBuilder.forIndexTask() - .withMaxRowsPerSegment(10000) + .withPartitionsSpec(new DynamicPartitionsSpec(10000, null)) .withMaxRowsInMemory(100) .withIndexSpec(indexSpec) .withMaxPendingPersists(3) @@ -749,7 +749,7 @@ public void testIndexTaskFailure() ), new IndexIOConfig(new MockExceptionInputSource(), new NoopInputFormat(), false, false), TuningConfigBuilder.forIndexTask() - .withMaxRowsPerSegment(10000) + .withPartitionsSpec(new DynamicPartitionsSpec(10000, null)) .withMaxRowsInMemory(10) .withIndexSpec(indexSpec) .withMaxPendingPersists(3) @@ -1179,7 +1179,7 @@ public void testResumeTasks() throws Exception ), new IndexIOConfig(new MockInputSource(), new NoopInputFormat(), false, false), TuningConfigBuilder.forIndexTask() - .withMaxRowsPerSegment(10000) + .withPartitionsSpec(new DynamicPartitionsSpec(10000, null)) .withMaxRowsInMemory(10) .withIndexSpec(indexSpec) .build() @@ -1235,7 +1235,6 @@ public void testUnifiedAppenderatorsManagerCleanup() throws Exception UnifiedIndexerAppenderatorsManager unifiedIndexerAppenderatorsManager = new UnifiedIndexerAppenderatorsManager( new ForwardingQueryProcessingPool(exec), - JoinableFactoryWrapperTest.NOOP_JOINABLE_FACTORY_WRAPPER, new WorkerConfig(), MapCache.create(2048), new CacheConfig(), @@ -1267,7 +1266,7 @@ public void testUnifiedAppenderatorsManagerCleanup() throws Exception ), new IndexIOConfig(new MockInputSource(), new NoopInputFormat(), false, false), TuningConfigBuilder.forIndexTask() - .withMaxRowsPerSegment(10000) + .withPartitionsSpec(new DynamicPartitionsSpec(10000, null)) .withMaxRowsInMemory(10) .withIndexSpec(indexSpec) .withMaxPendingPersists(3) diff --git a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/AppenderatorConfig.java b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/AppenderatorConfig.java index ab1b09b772cd..2e6652b2db06 100644 --- a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/AppenderatorConfig.java +++ b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/AppenderatorConfig.java @@ -42,15 +42,6 @@ default int getNumPersistThreads() return DEFAULT_NUM_PERSIST_THREADS; } - /** - * Maximum number of rows in a single segment before pushing to deep storage - */ - @Nullable - default Integer getMaxRowsPerSegment() - { - return Integer.MAX_VALUE; - } - /** * Maximum number of rows across all segments before pushing to deep storage */ diff --git a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManager.java b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManager.java index 0088e33ca7a8..0b880b7d9569 100644 --- a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManager.java +++ b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManager.java @@ -60,7 +60,6 @@ import org.apache.druid.segment.incremental.RowIngestionMeters; import org.apache.druid.segment.indexing.DataSchema; import org.apache.druid.segment.join.JoinableFactory; -import org.apache.druid.segment.join.JoinableFactoryWrapper; import org.apache.druid.segment.loading.DataSegmentPusher; import org.apache.druid.segment.loading.SegmentLoaderConfig; import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig; @@ -110,7 +109,6 @@ public class UnifiedIndexerAppenderatorsManager implements AppenderatorsManager private final Map datasourceBundles = new HashMap<>(); private final QueryProcessingPool queryProcessingPool; - private final JoinableFactoryWrapper joinableFactoryWrapper; private final WorkerConfig workerConfig; private final Cache cache; private final CacheConfig cacheConfig; @@ -124,7 +122,6 @@ public class UnifiedIndexerAppenderatorsManager implements AppenderatorsManager @Inject public UnifiedIndexerAppenderatorsManager( QueryProcessingPool queryProcessingPool, - JoinableFactoryWrapper joinableFactoryWrapper, WorkerConfig workerConfig, Cache cache, CacheConfig cacheConfig, @@ -135,7 +132,6 @@ public UnifiedIndexerAppenderatorsManager( ) { this.queryProcessingPool = queryProcessingPool; - this.joinableFactoryWrapper = joinableFactoryWrapper; this.workerConfig = workerConfig; this.cache = cache; this.cacheConfig = cacheConfig; @@ -440,13 +436,6 @@ public int getMaxPendingPersists() return baseConfig.getMaxPendingPersists(); } - @Nullable - @Override - public Integer getMaxRowsPerSegment() - { - return baseConfig.getMaxRowsPerSegment(); - } - @Nullable @Override public Long getMaxTotalRows() diff --git a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManagerTest.java b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManagerTest.java index 23ac93db0096..60f15bff4579 100644 --- a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManagerTest.java +++ b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManagerTest.java @@ -46,7 +46,6 @@ import org.apache.druid.segment.incremental.ParseExceptionHandler; import org.apache.druid.segment.indexing.DataSchema; import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec; -import org.apache.druid.segment.join.JoinableFactoryWrapperTest; import org.apache.druid.segment.loading.NoopDataSegmentPusher; import org.apache.druid.segment.metadata.CentralizedDatasourceSchemaConfig; import org.apache.druid.segment.realtime.SegmentGenerationMetrics; @@ -76,7 +75,6 @@ public class UnifiedIndexerAppenderatorsManagerTest extends InitializedNullHandl private final WorkerConfig workerConfig = new WorkerConfig(); private final UnifiedIndexerAppenderatorsManager manager = new UnifiedIndexerAppenderatorsManager( DirectQueryProcessingPool.INSTANCE, - JoinableFactoryWrapperTest.NOOP_JOINABLE_FACTORY_WRAPPER, workerConfig, MapCache.create(10), new CacheConfig(), diff --git a/services/src/test/java/org/apache/druid/cli/validate/DruidJsonValidatorTest.java b/services/src/test/java/org/apache/druid/cli/validate/DruidJsonValidatorTest.java index b617b7e6b877..e908279dfda9 100644 --- a/services/src/test/java/org/apache/druid/cli/validate/DruidJsonValidatorTest.java +++ b/services/src/test/java/org/apache/druid/cli/validate/DruidJsonValidatorTest.java @@ -149,16 +149,10 @@ public void testTaskValidator() throws Exception ), new IndexTask.IndexTuningConfig( - null, - null, null, 10, null, null, - null, - null, - null, - null, new DynamicPartitionsSpec(10000, null), IndexSpec.DEFAULT, null, @@ -172,7 +166,6 @@ public void testTaskValidator() throws Exception null, null, null, - null, 2 ) ), From 9b109fa8963a3897ad214009b1ff71dd5a3329a1 Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Mon, 29 Jul 2024 10:47:43 +0530 Subject: [PATCH 2/5] Fix usages --- .../org/apache/druid/k8s/overlord/common/K8sTestUtils.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/K8sTestUtils.java b/extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/K8sTestUtils.java index 3c91b8873e73..cf664cc99f74 100644 --- a/extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/K8sTestUtils.java +++ b/extensions-contrib/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/common/K8sTestUtils.java @@ -89,10 +89,6 @@ public static Task getTask() 10, null, null, - null, - 9999, - null, - null, new DynamicPartitionsSpec(10000, null), INDEX_SPEC, null, @@ -105,7 +101,6 @@ public static Task getTask() null, null, null, - null, 1L, null ) From 372bb2afa1379f8d1c6e7b7499f6974924fba9ae Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Mon, 29 Jul 2024 13:53:22 +0530 Subject: [PATCH 3/5] Remove references of maxRowsPerSegment in docs --- docs/api-reference/tasks-api.md | 5 ----- docs/ingestion/native-batch-simple-task.md | 2 -- docs/ingestion/native-batch.md | 4 ---- 3 files changed, 11 deletions(-) diff --git a/docs/api-reference/tasks-api.md b/docs/api-reference/tasks-api.md index d94be5b0c5fd..4a022ea72ce1 100644 --- a/docs/api-reference/tasks-api.md +++ b/docs/api-reference/tasks-api.md @@ -756,7 +756,6 @@ Host: http://ROUTER_IP:ROUTER_PORT }, "tuningConfig": { "type": "index_parallel", - "maxRowsPerSegment": 5000000, "appendableIndexSpec": { "type": "onheap", "preserveExistingMetrics": false @@ -764,8 +763,6 @@ Host: http://ROUTER_IP:ROUTER_PORT "maxRowsInMemory": 25000, "maxBytesInMemory": 0, "skipBytesInMemoryOverheadCheck": false, - "maxTotalRows": null, - "numShards": null, "splitHintSpec": null, "partitionsSpec": { "type": "dynamic", @@ -1282,7 +1279,6 @@ curl "http://ROUTER_IP:ROUTER_PORT/druid/indexer/v1/task" \ }, "tuningConfig" : { "type" : "index_parallel", - "maxRowsPerSegment" : 5000000, "maxRowsInMemory" : 25000 } } @@ -1334,7 +1330,6 @@ Content-Length: 952 }, "tuningConfig" : { "type" : "index_parallel", - "maxRowsPerSegment" : 5000000, "maxRowsInMemory" : 25000 } } diff --git a/docs/ingestion/native-batch-simple-task.md b/docs/ingestion/native-batch-simple-task.md index 6b93d119ec87..57964d342220 100644 --- a/docs/ingestion/native-batch-simple-task.md +++ b/docs/ingestion/native-batch-simple-task.md @@ -138,8 +138,6 @@ The tuningConfig is optional and default parameters will be used if no tuningCon |type|The task type, this should always be "index".|none|yes| |maxRowsInMemory|Used in determining when intermediate persists to disk should occur. Normally user does not need to set this, but depending on the nature of data, if rows are short in terms of bytes, user may not want to store a million rows in memory and this value should be set.|1000000|no| |maxBytesInMemory|Used in determining when intermediate persists to disk should occur. Normally this is computed internally and user does not need to set it. This value represents number of bytes to aggregate in heap memory before persisting. This is based on a rough estimate of memory usage and not actual usage. The maximum heap memory usage for indexing is maxBytesInMemory * (2 + maxPendingPersists). Note that `maxBytesInMemory` also includes heap usage of artifacts created from intermediary persists. This means that after every persist, the amount of `maxBytesInMemory` until next persist will decreases, and task will fail when the sum of bytes of all intermediary persisted artifacts exceeds `maxBytesInMemory`.|1/6 of max JVM memory|no| -|maxTotalRows|Deprecated. Use `partitionsSpec` instead. Total number of rows in segments waiting for being pushed. Used in determining when intermediate pushing should occur.|20000000|no| -|numShards|Deprecated. Use `partitionsSpec` instead. Directly specify the number of shards to create. If this is specified and `intervals` is specified in the `granularitySpec`, the index task can skip the determine intervals/partitions pass through the data.|null|no| |partitionDimensions|Deprecated. Use `partitionsSpec` instead. The dimensions to partition on. Leave blank to select all dimensions. Only used with `forceGuaranteedRollup` = true, will be ignored otherwise.|null|no| |partitionsSpec|Defines how to partition data in each timeChunk, see [PartitionsSpec](#partitionsspec)|`dynamic` if `forceGuaranteedRollup` = false, `hashed` if `forceGuaranteedRollup` = true|no| |indexSpec|Defines segment storage format options to be used at indexing time, see [IndexSpec](ingestion-spec.md#indexspec)|null|no| diff --git a/docs/ingestion/native-batch.md b/docs/ingestion/native-batch.md index 398fea9f69a9..6bf37a7b7676 100644 --- a/docs/ingestion/native-batch.md +++ b/docs/ingestion/native-batch.md @@ -261,8 +261,6 @@ The following table lists the properties of a `tuningConfig` object: |`maxRowsInMemory`|Determines when Druid should perform intermediate persists to disk. Normally you don't need to set this. Depending on the nature of your data, if rows are short in terms of bytes. For example, you may not want to store a million rows in memory. In this case, set this value.|1000000|no| |`maxBytesInMemory`|Use to determine when Druid should perform intermediate persists to disk. Normally Druid computes this internally and you don't need to set it. This value represents number of bytes to aggregate in heap memory before persisting. This is based on a rough estimate of memory usage and not actual usage. The maximum heap memory usage for indexing is `maxBytesInMemory * (2 + maxPendingPersists)`. Note that `maxBytesInMemory` also includes heap usage of artifacts created from intermediary persists. This means that after every persist, the amount of `maxBytesInMemory` until next persist will decrease. Tasks fail when the sum of bytes of all intermediary persisted artifacts exceeds `maxBytesInMemory`.|1/6 of max JVM memory|no| |`maxColumnsToMerge`|Limit of the number of segments to merge in a single phase when merging segments for publishing. This limit affects the total number of columns present in a set of segments to merge. If the limit is exceeded, segment merging occurs in multiple phases. Druid merges at least 2 segments per phase, regardless of this setting.|-1 (unlimited)|no| -|`maxTotalRows`|Deprecated. Use `partitionsSpec` instead. Total number of rows in segments waiting to be pushed. Used to determine when intermediate pushing should occur.|20000000|no| -|`numShards`|Deprecated. Use `partitionsSpec` instead. Directly specify the number of shards to create when using a `hashed` `partitionsSpec`. If this is specified and `intervals` is specified in the `granularitySpec`, the index task can skip the determine intervals/partitions pass through the data.|null|no| |`splitHintSpec`|Hint to control the amount of data that each first phase task reads. Druid may ignore the hint depending on the implementation of the input source. See [Split hint spec](#split-hint-spec) for more details.|size-based split hint spec|no| |`partitionsSpec`|Defines how to partition data in each timeChunk, see [PartitionsSpec](#partitionsspec).|`dynamic` if `forceGuaranteedRollup` = false, `hashed` or `single_dim` if `forceGuaranteedRollup` = true|no| |`indexSpec`|Defines segment storage format options to be used at indexing time, see [IndexSpec](ingestion-spec.md#indexspec).|null|no| @@ -659,8 +657,6 @@ The returned result contains the worker task spec, a current task status if exis "type": "dynamic" }, "maxRowsInMemory": 1000000, - "maxTotalRows": 20000000, - "numShards": null, "indexSpec": { "bitmap": { "type": "roaring" From 260195e93d6eefe721b84bf15887f31b4cbe6a07 Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Mon, 29 Jul 2024 19:15:36 +0530 Subject: [PATCH 4/5] Fix tests --- .../indexing/common/task/TaskSerdeTest.java | 88 +++++++------------ 1 file changed, 33 insertions(+), 55 deletions(-) diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskSerdeTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskSerdeTest.java index e5a03a9c7f53..6f0c80575971 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskSerdeTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/TaskSerdeTest.java @@ -29,8 +29,9 @@ import org.apache.druid.data.input.impl.TimestampSpec; import org.apache.druid.indexer.HadoopIOConfig; import org.apache.druid.indexer.HadoopIngestionSpec; +import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec; import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; -import org.apache.druid.indexer.partitions.PartitionsSpec; +import org.apache.druid.indexer.partitions.HashedPartitionsSpec; import org.apache.druid.indexing.common.TestUtils; import org.apache.druid.indexing.common.task.IndexTask.IndexIOConfig; import org.apache.druid.indexing.common.task.IndexTask.IndexIngestionSpec; @@ -52,6 +53,7 @@ import org.junit.rules.ExpectedException; import java.io.File; +import java.util.Arrays; public class TaskSerdeTest { @@ -87,7 +89,7 @@ public void testIndexTaskIOConfigDefaults() throws Exception public void testIndexTaskTuningConfigDefaults() throws Exception { final IndexTask.IndexTuningConfig tuningConfig = jsonMapper.readValue( - "{\"type\":\"index\"}", + json("{'type': 'index'}"), IndexTask.IndexTuningConfig.class ); @@ -96,81 +98,53 @@ public void testIndexTaskTuningConfigDefaults() throws Exception Assert.assertEquals(new Period(Integer.MAX_VALUE), tuningConfig.getIntermediatePersistPeriod()); Assert.assertEquals(0, tuningConfig.getMaxPendingPersists()); Assert.assertEquals(1000000, tuningConfig.getMaxRowsInMemory()); - // TODO: Assert.assertNull(tuningConfig.getNumShards()); - Assert.assertNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); + Assert.assertEquals(new DynamicPartitionsSpec(null, null), tuningConfig.getGivenOrDefaultPartitionsSpec()); } @Test - public void testIndexTaskTuningConfigTargetPartitionSizeOrNumShards() throws Exception + public void testIndexTaskTuningConfigPartitionsSpec() throws Exception { IndexTask.IndexTuningConfig tuningConfig = jsonMapper.readValue( - "{\"type\":\"index\", \"targetPartitionSize\":10}", + json("{'type': 'index', 'partitionsSpec':{'type': 'dynamic'}}"), IndexTask.IndexTuningConfig.class ); - - Assert.assertEquals(10, (int) tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); - // Assert.assertNull(tuningConfig.getNumShards()); - - tuningConfig = jsonMapper.readValue( - "{\"type\":\"index\"}", - IndexTask.IndexTuningConfig.class + Assert.assertEquals( + new DynamicPartitionsSpec(null, null), + tuningConfig.getGivenOrDefaultPartitionsSpec() ); - Assert.assertNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); - tuningConfig = jsonMapper.readValue( - "{\"type\":\"index\", \"maxRowsPerSegment\":10}", + json("{'type': 'index', 'partitionsSpec':{'type': 'dynamic', 'maxRowsPerSegment':100}}"), IndexTask.IndexTuningConfig.class ); - - Assert.assertEquals(10, (int) tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); - // Assert.assertNull(tuningConfig.getNumShards()); - - tuningConfig = jsonMapper.readValue( - "{\"type\":\"index\", \"numShards\":10, \"forceGuaranteedRollup\": true}", - IndexTask.IndexTuningConfig.class + Assert.assertEquals( + new DynamicPartitionsSpec(100, null), + tuningConfig.getGivenOrDefaultPartitionsSpec() ); - Assert.assertNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); - // Assert.assertEquals(10, (int) tuningConfig.getNumShards()); - tuningConfig = jsonMapper.readValue( - "{\"type\":\"index\", \"targetPartitionSize\":-1, \"numShards\":10, \"forceGuaranteedRollup\": true}", + json("{'type': 'index', 'forceGuaranteedRollup': true," + + " 'partitionsSpec':{'type': 'hashed', 'numShards':100}}"), IndexTask.IndexTuningConfig.class ); - - Assert.assertNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); - // Assert.assertEquals(10, (int) tuningConfig.getNumShards()); - - tuningConfig = jsonMapper.readValue( - "{\"type\":\"index\", \"targetPartitionSize\":10, \"numShards\":-1}", - IndexTask.IndexTuningConfig.class + Assert.assertEquals( + new HashedPartitionsSpec(null, 100, null), + tuningConfig.getGivenOrDefaultPartitionsSpec() ); - // Assert.assertNull(tuningConfig.getNumShards()); - Assert.assertEquals(10, (int) tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); - tuningConfig = jsonMapper.readValue( - "{\"type\":\"index\", \"targetPartitionSize\":-1, \"numShards\":-1, \"forceGuaranteedRollup\": true}", + json( + "{" + + "'type': 'index'," + + " 'forceGuaranteedRollup': true," + + " 'partitionsSpec':{'type': 'range', 'partitionDimensions':['d1', 'd2'], 'maxRowsPerSegment': 100}" + + "}" + ), IndexTask.IndexTuningConfig.class ); - - // Assert.assertNull(tuningConfig.getNumShards()); - Assert.assertNotNull(tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment()); Assert.assertEquals( - PartitionsSpec.DEFAULT_MAX_ROWS_PER_SEGMENT, - tuningConfig.getGivenOrDefaultPartitionsSpec().getMaxRowsPerSegment().intValue() - ); - } - - @Test - public void testIndexTaskTuningConfigTargetPartitionSizeAndNumShards() throws Exception - { - thrown.expectCause(CoreMatchers.isA(IllegalArgumentException.class)); - - jsonMapper.readValue( - "{\"type\":\"index\", \"targetPartitionSize\":10, \"numShards\":10, \"forceGuaranteedRollup\": true}", - IndexTask.IndexTuningConfig.class + new DimensionRangePartitionsSpec(null, 100, Arrays.asList("d1", "d2"), false), + tuningConfig.getGivenOrDefaultPartitionsSpec() ); } @@ -278,7 +252,6 @@ public void testIndexTaskSerde() throws Exception ); Assert.assertEquals(taskTuningConfig.getMaxPendingPersists(), task2TuningConfig.getMaxPendingPersists()); Assert.assertEquals(taskTuningConfig.getMaxRowsInMemory(), task2TuningConfig.getMaxRowsInMemory()); - // Assert.assertEquals(taskTuningConfig.getNumShards(), task2TuningConfig.getNumShards()); Assert.assertEquals( taskTuningConfig.getGivenOrDefaultPartitionsSpec(), task2TuningConfig.getGivenOrDefaultPartitionsSpec() @@ -501,4 +474,9 @@ public void testHadoopIndexTaskWithContextSerde() throws Exception Assert.assertEquals(ImmutableMap.of("userid", 12345, "username", "bob"), task2.getContext()); Assert.assertEquals(ImmutableMap.of("userid", 12345, "username", "bob"), task2.getSpec().getContext()); } + + private static String json(String singleQuotedJson) + { + return TestUtils.singleQuoteToStandardJson(singleQuotedJson); + } } From 7bc4a4848e1d0f932d586e1999e15293ff9cc9c4 Mon Sep 17 00:00:00 2001 From: Kashif Faraz Date: Tue, 30 Jul 2024 09:07:49 +0530 Subject: [PATCH 5/5] Fix tests, clean up DataSourceCompactionConfig --- .../ClientCompactionTaskQuerySerdeTest.java | 2 - .../druid/testing/utils/CompactionUtil.java | 2 - .../duty/ITAutoCompactionTest.java | 2 - .../duty/ITAutoCompactionUpgradeTest.java | 2 - ...ClientCompactionTaskQueryTuningConfig.java | 45 ++-------------- .../DataSourceCompactionConfig.java | 18 ------- .../UserCompactionTaskQueryTuningConfig.java | 12 ----- .../coordinator/compact/CompactionStatus.java | 3 +- .../coordinator/duty/CompactSegments.java | 1 - .../ClientCompactionRunnerInfoTest.java | 8 +-- .../DataSourceCompactionConfigTest.java | 39 +------------- ...erCompactionTaskQueryTuningConfigTest.java | 2 - .../compact/CompactionStatusTest.java | 52 ++----------------- .../compact/NewestSegmentFirstPolicyTest.java | 6 +-- .../coordinator/duty/CompactSegmentsTest.java | 32 ------------ .../duty/KillCompactionConfigTest.java | 3 -- ...rdinatorCompactionConfigsResourceTest.java | 7 --- 17 files changed, 13 insertions(+), 223 deletions(-) diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java index 55898357d663..9d93d2cc3c6d 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/ClientCompactionTaskQuerySerdeTest.java @@ -304,11 +304,9 @@ private ClientCompactionTaskQuery createCompactionTaskQuery(String id, ClientCom new ClientCompactionIntervalSpec(Intervals.of("2019/2020"), "testSha256OfSortedSegmentIds"), true ), new ClientCompactionTaskQueryTuningConfig( - 100, new OnheapIncrementalIndex.Spec(true), 40000, 2000L, - 30000L, SEGMENTS_SPLIT_HINT_SPEC, DYNAMIC_PARTITIONS_SPEC, INDEX_SPEC, diff --git a/integration-tests/src/main/java/org/apache/druid/testing/utils/CompactionUtil.java b/integration-tests/src/main/java/org/apache/druid/testing/utils/CompactionUtil.java index 63316e76687d..b61bba96217d 100644 --- a/integration-tests/src/main/java/org/apache/druid/testing/utils/CompactionUtil.java +++ b/integration-tests/src/main/java/org/apache/druid/testing/utils/CompactionUtil.java @@ -47,13 +47,11 @@ public static DataSourceCompactionConfig createCompactionConfig( fullDatasourceName, null, null, - null, skipOffsetFromLatest, new UserCompactionTaskQueryTuningConfig( null, null, null, - null, new MaxSizeSplitHintSpec(null, 1), new DynamicPartitionsSpec(maxRowsPerSegment, null), null, diff --git a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java b/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java index 230a19236c16..4e90fe77a189 100644 --- a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java +++ b/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionTest.java @@ -1800,13 +1800,11 @@ private void submitCompactionConfig( fullDatasourceName, null, null, - null, skipOffsetFromLatest, new UserCompactionTaskQueryTuningConfig( null, null, null, - null, new MaxSizeSplitHintSpec(null, 1), partitionsSpec, null, diff --git a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionUpgradeTest.java b/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionUpgradeTest.java index 9e1b54143bbd..8f4524794010 100644 --- a/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionUpgradeTest.java +++ b/integration-tests/src/test/java/org/apache/druid/tests/coordinator/duty/ITAutoCompactionUpgradeTest.java @@ -75,13 +75,11 @@ public void testUpgradeAutoCompactionConfigurationWhenConfigurationFromOlderVers UPGRADE_DATASOURCE_NAME, null, null, - null, newSkipOffset, new UserCompactionTaskQueryTuningConfig( null, null, null, - null, new MaxSizeSplitHintSpec(null, 1), newPartitionsSpec, null, diff --git a/server/src/main/java/org/apache/druid/client/indexing/ClientCompactionTaskQueryTuningConfig.java b/server/src/main/java/org/apache/druid/client/indexing/ClientCompactionTaskQueryTuningConfig.java index 55fe7d0114f9..82d7c18695d0 100644 --- a/server/src/main/java/org/apache/druid/client/indexing/ClientCompactionTaskQueryTuningConfig.java +++ b/server/src/main/java/org/apache/druid/client/indexing/ClientCompactionTaskQueryTuningConfig.java @@ -37,16 +37,10 @@ public class ClientCompactionTaskQueryTuningConfig { - @Deprecated - @Nullable - private final Integer maxRowsPerSegment; @Nullable private final Integer maxRowsInMemory; @Nullable private final Long maxBytesInMemory; - @Deprecated - @Nullable - private final Long maxTotalRows; @Nullable private final SplitHintSpec splitHintSpec; @Nullable @@ -85,21 +79,19 @@ public static ClientCompactionTaskQueryTuningConfig from( ) { if (compactionConfig == null) { - return from(null, null, null); + return from(null, null); } else { - return from(compactionConfig.getTuningConfig(), compactionConfig.getMaxRowsPerSegment(), null); + return from(compactionConfig.getTuningConfig(), null); } } public static ClientCompactionTaskQueryTuningConfig from( @Nullable UserCompactionTaskQueryTuningConfig userCompactionTaskQueryTuningConfig, - @Nullable Integer maxRowsPerSegment, @Nullable Boolean preserveExistingMetrics ) { if (userCompactionTaskQueryTuningConfig == null) { return new ClientCompactionTaskQueryTuningConfig( - maxRowsPerSegment, new OnheapIncrementalIndex.Spec(preserveExistingMetrics), null, null, @@ -117,7 +109,6 @@ public static ClientCompactionTaskQueryTuningConfig from( null, null, null, - null, null ); } else { @@ -125,11 +116,9 @@ public static ClientCompactionTaskQueryTuningConfig from( ? userCompactionTaskQueryTuningConfig.getAppendableIndexSpec() : new OnheapIncrementalIndex.Spec(preserveExistingMetrics); return new ClientCompactionTaskQueryTuningConfig( - maxRowsPerSegment, appendableIndexSpecToUse, userCompactionTaskQueryTuningConfig.getMaxRowsInMemory(), userCompactionTaskQueryTuningConfig.getMaxBytesInMemory(), - userCompactionTaskQueryTuningConfig.getMaxTotalRows(), userCompactionTaskQueryTuningConfig.getSplitHintSpec(), userCompactionTaskQueryTuningConfig.getPartitionsSpec(), userCompactionTaskQueryTuningConfig.getIndexSpec(), @@ -151,11 +140,9 @@ public static ClientCompactionTaskQueryTuningConfig from( @JsonCreator public ClientCompactionTaskQueryTuningConfig( - @JsonProperty("maxRowsPerSegment") @Deprecated @Nullable Integer maxRowsPerSegment, @JsonProperty("appendableIndexSpec") @Nullable AppendableIndexSpec appendableIndexSpec, @JsonProperty("maxRowsInMemory") @Nullable Integer maxRowsInMemory, @JsonProperty("maxBytesInMemory") @Nullable Long maxBytesInMemory, - @JsonProperty("maxTotalRows") @Deprecated @Nullable Long maxTotalRows, @JsonProperty("splitHintSpec") @Nullable SplitHintSpec splitHintSpec, @JsonProperty("partitionsSpec") @Nullable PartitionsSpec partitionsSpec, @JsonProperty("indexSpec") @Nullable IndexSpec indexSpec, @@ -173,11 +160,9 @@ public ClientCompactionTaskQueryTuningConfig( @JsonProperty("maxColumnsToMerge") @Nullable Integer maxColumnsToMerge ) { - this.maxRowsPerSegment = maxRowsPerSegment; this.appendableIndexSpec = appendableIndexSpec; this.maxRowsInMemory = maxRowsInMemory; this.maxBytesInMemory = maxBytesInMemory; - this.maxTotalRows = maxTotalRows; this.splitHintSpec = splitHintSpec; this.partitionsSpec = partitionsSpec; this.indexSpec = indexSpec; @@ -201,14 +186,6 @@ public String getType() return "index_parallel"; } - @Deprecated - @JsonProperty - @Nullable - public Integer getMaxRowsPerSegment() - { - return maxRowsPerSegment; - } - @JsonProperty @Nullable public Integer getMaxRowsInMemory() @@ -223,14 +200,6 @@ public Long getMaxBytesInMemory() return maxBytesInMemory; } - @Deprecated - @JsonProperty - @Nullable - public Long getMaxTotalRows() - { - return maxTotalRows; - } - @JsonProperty @Nullable public SplitHintSpec getSplitHintSpec() @@ -360,10 +329,8 @@ public boolean equals(Object o) return false; } ClientCompactionTaskQueryTuningConfig that = (ClientCompactionTaskQueryTuningConfig) o; - return Objects.equals(maxRowsPerSegment, that.maxRowsPerSegment) && - Objects.equals(maxRowsInMemory, that.maxRowsInMemory) && + return Objects.equals(maxRowsInMemory, that.maxRowsInMemory) && Objects.equals(maxBytesInMemory, that.maxBytesInMemory) && - Objects.equals(maxTotalRows, that.maxTotalRows) && Objects.equals(splitHintSpec, that.splitHintSpec) && Objects.equals(partitionsSpec, that.partitionsSpec) && Objects.equals(indexSpec, that.indexSpec) && @@ -386,10 +353,8 @@ public boolean equals(Object o) public int hashCode() { return Objects.hash( - maxRowsPerSegment, maxRowsInMemory, maxBytesInMemory, - maxTotalRows, splitHintSpec, partitionsSpec, indexSpec, @@ -413,10 +378,8 @@ public int hashCode() public String toString() { return "ClientCompactionTaskQueryTuningConfig{" + - "maxRowsPerSegment=" + maxRowsPerSegment + - ", maxRowsInMemory=" + maxRowsInMemory + + "maxRowsInMemory=" + maxRowsInMemory + ", maxBytesInMemory=" + maxBytesInMemory + - ", maxTotalRows=" + maxTotalRows + ", splitHintSpec=" + splitHintSpec + ", partitionsSpec=" + partitionsSpec + ", indexSpec=" + indexSpec + diff --git a/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java index 767e8218f319..47952ac8c92f 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/DataSourceCompactionConfig.java @@ -42,12 +42,6 @@ public class DataSourceCompactionConfig private final String dataSource; private final int taskPriority; private final long inputSegmentSizeBytes; - /** - * The number of input segments is limited because the byte size of a serialized task spec is limited by - * org.apache.druid.indexing.overlord.config.RemoteTaskRunnerConfig.maxZnodeBytes. - */ - @Nullable - private final Integer maxRowsPerSegment; private final Period skipOffsetFromLatest; private final UserCompactionTaskQueryTuningConfig tuningConfig; private final UserCompactionTaskGranularityConfig granularitySpec; @@ -63,7 +57,6 @@ public DataSourceCompactionConfig( @JsonProperty("dataSource") String dataSource, @JsonProperty("taskPriority") @Nullable Integer taskPriority, @JsonProperty("inputSegmentSizeBytes") @Nullable Long inputSegmentSizeBytes, - @JsonProperty("maxRowsPerSegment") @Deprecated @Nullable Integer maxRowsPerSegment, @JsonProperty("skipOffsetFromLatest") @Nullable Period skipOffsetFromLatest, @JsonProperty("tuningConfig") @Nullable UserCompactionTaskQueryTuningConfig tuningConfig, @JsonProperty("granularitySpec") @Nullable UserCompactionTaskGranularityConfig granularitySpec, @@ -82,7 +75,6 @@ public DataSourceCompactionConfig( this.inputSegmentSizeBytes = inputSegmentSizeBytes == null ? DEFAULT_INPUT_SEGMENT_SIZE_BYTES : inputSegmentSizeBytes; - this.maxRowsPerSegment = maxRowsPerSegment; this.skipOffsetFromLatest = skipOffsetFromLatest == null ? DEFAULT_SKIP_OFFSET_FROM_LATEST : skipOffsetFromLatest; this.tuningConfig = tuningConfig; this.ioConfig = ioConfig; @@ -112,14 +104,6 @@ public long getInputSegmentSizeBytes() return inputSegmentSizeBytes; } - @Deprecated - @JsonProperty - @Nullable - public Integer getMaxRowsPerSegment() - { - return maxRowsPerSegment; - } - @JsonProperty public Period getSkipOffsetFromLatest() { @@ -195,7 +179,6 @@ public boolean equals(Object o) return taskPriority == that.taskPriority && inputSegmentSizeBytes == that.inputSegmentSizeBytes && Objects.equals(dataSource, that.dataSource) && - Objects.equals(maxRowsPerSegment, that.maxRowsPerSegment) && Objects.equals(skipOffsetFromLatest, that.skipOffsetFromLatest) && Objects.equals(tuningConfig, that.tuningConfig) && Objects.equals(granularitySpec, that.granularitySpec) && @@ -214,7 +197,6 @@ public int hashCode() dataSource, taskPriority, inputSegmentSizeBytes, - maxRowsPerSegment, skipOffsetFromLatest, tuningConfig, granularitySpec, diff --git a/server/src/main/java/org/apache/druid/server/coordinator/UserCompactionTaskQueryTuningConfig.java b/server/src/main/java/org/apache/druid/server/coordinator/UserCompactionTaskQueryTuningConfig.java index 9c1bfb200648..4de8369d5c6a 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/UserCompactionTaskQueryTuningConfig.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/UserCompactionTaskQueryTuningConfig.java @@ -20,7 +20,6 @@ package org.apache.druid.server.coordinator; import com.fasterxml.jackson.annotation.JsonCreator; -import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonProperty; import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig; import org.apache.druid.data.input.SplitHintSpec; @@ -39,7 +38,6 @@ public UserCompactionTaskQueryTuningConfig( @JsonProperty("maxRowsInMemory") @Nullable Integer maxRowsInMemory, @JsonProperty("appendableIndexSpec") @Nullable AppendableIndexSpec appendableIndexSpec, @JsonProperty("maxBytesInMemory") @Nullable Long maxBytesInMemory, - @JsonProperty("maxTotalRows") @Deprecated @Nullable Long maxTotalRows, @JsonProperty("splitHintSpec") @Nullable SplitHintSpec splitHintSpec, @JsonProperty("partitionsSpec") @Nullable PartitionsSpec partitionsSpec, @JsonProperty("indexSpec") @Nullable IndexSpec indexSpec, @@ -58,11 +56,9 @@ public UserCompactionTaskQueryTuningConfig( ) { super( - null, appendableIndexSpec, maxRowsInMemory, maxBytesInMemory, - maxTotalRows, splitHintSpec, partitionsSpec, indexSpec, @@ -80,12 +76,4 @@ public UserCompactionTaskQueryTuningConfig( maxColumnsToMerge ); } - - @Override - @Nullable - @JsonIgnore - public Integer getMaxRowsPerSegment() - { - throw new UnsupportedOperationException(); - } } diff --git a/server/src/main/java/org/apache/druid/server/coordinator/compact/CompactionStatus.java b/server/src/main/java/org/apache/druid/server/coordinator/compact/CompactionStatus.java index fa053fb8d6ad..b826deac0f1f 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/compact/CompactionStatus.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/compact/CompactionStatus.java @@ -130,8 +130,7 @@ static PartitionsSpec findPartitionsSpecFromConfig(ClientCompactionTaskQueryTuni { final PartitionsSpec partitionsSpecFromTuningConfig = tuningConfig.getPartitionsSpec(); if (partitionsSpecFromTuningConfig == null) { - final long maxTotalRows = Configs.valueOrDefault(tuningConfig.getMaxTotalRows(), Long.MAX_VALUE); - return new DynamicPartitionsSpec(tuningConfig.getMaxRowsPerSegment(), maxTotalRows); + return new DynamicPartitionsSpec(null, null); } else if (partitionsSpecFromTuningConfig instanceof DynamicPartitionsSpec) { return new DynamicPartitionsSpec( partitionsSpecFromTuningConfig.getMaxRowsPerSegment(), diff --git a/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java b/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java index 01f3bc77e9ee..e68e1517687e 100644 --- a/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java +++ b/server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java @@ -509,7 +509,6 @@ private int submitCompactionTasks( config.getTaskPriority(), ClientCompactionTaskQueryTuningConfig.from( config.getTuningConfig(), - config.getMaxRowsPerSegment(), config.getMetricsSpec() != null ), granularitySpec, diff --git a/server/src/test/java/org/apache/druid/client/indexing/ClientCompactionRunnerInfoTest.java b/server/src/test/java/org/apache/druid/client/indexing/ClientCompactionRunnerInfoTest.java index f6d4a2b6e581..40edbc5e0fd3 100644 --- a/server/src/test/java/org/apache/druid/client/indexing/ClientCompactionRunnerInfoTest.java +++ b/server/src/test/java/org/apache/druid/client/indexing/ClientCompactionRunnerInfoTest.java @@ -192,11 +192,10 @@ private static DataSourceCompactionConfig createCompactionConfig( @Nullable AggregatorFactory[] metricsSpec ) { - final DataSourceCompactionConfig config = new DataSourceCompactionConfig( + return new DataSourceCompactionConfig( "dataSource", null, 500L, - 10000, new Period(3600), createTuningConfig(partitionsSpec), granularitySpec, @@ -207,15 +206,13 @@ private static DataSourceCompactionConfig createCompactionConfig( CompactionEngine.MSQ, context ); - return config; } private static UserCompactionTaskQueryTuningConfig createTuningConfig(PartitionsSpec partitionsSpec) { - final UserCompactionTaskQueryTuningConfig tuningConfig = new UserCompactionTaskQueryTuningConfig( + return new UserCompactionTaskQueryTuningConfig( 40000, null, - 2000L, null, new SegmentsSplitHintSpec(new HumanReadableBytes(100000L), null), partitionsSpec, @@ -241,6 +238,5 @@ private static UserCompactionTaskQueryTuningConfig createTuningConfig(Partitions 100, 2 ); - return tuningConfig; } } diff --git a/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigTest.java index a9334f077a47..8f27ffc5142a 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/DataSourceCompactionConfigTest.java @@ -42,9 +42,7 @@ import org.joda.time.Duration; import org.joda.time.Period; import org.junit.Assert; -import org.junit.Rule; import org.junit.Test; -import org.junit.rules.ExpectedException; import java.io.IOException; @@ -52,9 +50,6 @@ public class DataSourceCompactionConfigTest extends InitializedNullHandlingTest { private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper(); - @Rule - public final ExpectedException expectedException = ExpectedException.none(); - @Test public void testSerdeBasic() throws IOException { @@ -62,7 +57,6 @@ public void testSerdeBasic() throws IOException "dataSource", null, null, - null, new Period(3600), null, null, @@ -79,7 +73,6 @@ public void testSerdeBasic() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(100_000_000_000_000L, fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -94,7 +87,6 @@ public void testSerdeWithMaxRowsPerSegment() throws IOException "dataSource", null, 500L, - 30, new Period(3600), null, null, @@ -111,7 +103,6 @@ public void testSerdeWithMaxRowsPerSegment() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -125,13 +116,11 @@ public void testSerdeWithMaxTotalRows() throws IOException "dataSource", null, 500L, - null, new Period(3600), new UserCompactionTaskQueryTuningConfig( null, null, null, - 10000L, null, null, null, @@ -162,7 +151,6 @@ public void testSerdeWithMaxTotalRows() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -176,13 +164,11 @@ public void testSerdeMaxTotalRowsWithMaxRowsPerSegment() throws IOException "dataSource", null, 500L, - 10000, new Period(3600), new UserCompactionTaskQueryTuningConfig( null, null, null, - 10000L, null, null, null, @@ -214,7 +200,6 @@ public void testSerdeMaxTotalRowsWithMaxRowsPerSegment() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -226,7 +211,6 @@ public void testSerdeUserCompactionTuningConfig() throws IOException final UserCompactionTaskQueryTuningConfig tuningConfig = new UserCompactionTaskQueryTuningConfig( 40000, null, - 2000L, null, new SegmentsSplitHintSpec(new HumanReadableBytes(100000L), null), new DynamicPartitionsSpec(1000, 20000L), @@ -265,7 +249,6 @@ public void testSerdeUserCompactionTuningConfigWithAppendableIndexSpec() throws final UserCompactionTaskQueryTuningConfig tuningConfig = new UserCompactionTaskQueryTuningConfig( 40000, new OnheapIncrementalIndex.Spec(true), - 2000L, null, new SegmentsSplitHintSpec(new HumanReadableBytes(100000L), null), new DynamicPartitionsSpec(1000, 20000L), @@ -305,7 +288,6 @@ public void testSerdeGranularitySpec() throws IOException "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), @@ -322,7 +304,6 @@ public void testSerdeGranularitySpec() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -336,7 +317,6 @@ public void testSerdeGranularitySpecWithQueryGranularity() throws Exception "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(null, Granularities.YEAR, null), @@ -353,7 +333,6 @@ public void testSerdeGranularitySpecWithQueryGranularity() throws Exception Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -370,7 +349,6 @@ public void testSerdeWithNullGranularitySpec() throws IOException "dataSource", null, 500L, - null, new Period(3600), null, null, @@ -387,7 +365,6 @@ public void testSerdeWithNullGranularitySpec() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -401,7 +378,6 @@ public void testSerdeGranularitySpecWithNullValues() throws IOException "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(null, null, null), @@ -418,7 +394,6 @@ public void testSerdeGranularitySpecWithNullValues() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -432,7 +407,6 @@ public void testSerdeGranularitySpecWithRollup() throws IOException "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(null, null, true), @@ -449,7 +423,6 @@ public void testSerdeGranularitySpecWithRollup() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -466,7 +439,6 @@ public void testSerdeIOConfigWithNonNullDropExisting() throws IOException "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), @@ -483,7 +455,6 @@ public void testSerdeIOConfigWithNonNullDropExisting() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -498,7 +469,6 @@ public void testSerdeIOConfigWithNullDropExisting() throws IOException "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), @@ -515,7 +485,6 @@ public void testSerdeIOConfigWithNullDropExisting() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -530,7 +499,6 @@ public void testSerdeDimensionsSpec() throws IOException "dataSource", null, 500L, - null, new Period(3600), null, null, @@ -547,7 +515,6 @@ public void testSerdeDimensionsSpec() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -562,7 +529,6 @@ public void testSerdeTransformSpec() throws IOException "dataSource", null, 500L, - null, new Period(3600), null, null, @@ -579,7 +545,6 @@ public void testSerdeTransformSpec() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); @@ -593,7 +558,6 @@ public void testSerdeMetricsSpec() throws IOException "dataSource", null, 500L, - null, new Period(3600), null, null, @@ -610,10 +574,9 @@ public void testSerdeMetricsSpec() throws IOException Assert.assertEquals(config.getDataSource(), fromJson.getDataSource()); Assert.assertEquals(25, fromJson.getTaskPriority()); Assert.assertEquals(config.getInputSegmentSizeBytes(), fromJson.getInputSegmentSizeBytes()); - Assert.assertEquals(config.getMaxRowsPerSegment(), fromJson.getMaxRowsPerSegment()); Assert.assertEquals(config.getSkipOffsetFromLatest(), fromJson.getSkipOffsetFromLatest()); Assert.assertEquals(config.getTuningConfig(), fromJson.getTuningConfig()); Assert.assertEquals(config.getTaskContext(), fromJson.getTaskContext()); - Assert.assertEquals(config.getMetricsSpec(), fromJson.getMetricsSpec()); + Assert.assertArrayEquals(config.getMetricsSpec(), fromJson.getMetricsSpec()); } } diff --git a/server/src/test/java/org/apache/druid/server/coordinator/UserCompactionTaskQueryTuningConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/UserCompactionTaskQueryTuningConfigTest.java index a92122e475d7..e74bc7215a30 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/UserCompactionTaskQueryTuningConfigTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/UserCompactionTaskQueryTuningConfigTest.java @@ -61,7 +61,6 @@ public void testSerdeNulls() throws IOException null, null, null, - null, null ); final String json = OBJECT_MAPPER.writeValueAsString(config); @@ -79,7 +78,6 @@ public void testSerde() throws IOException 40000, new OnheapIncrementalIndex.Spec(true), 2000L, - null, new SegmentsSplitHintSpec(new HumanReadableBytes(42L), null), new DynamicPartitionsSpec(1000, 20000L), IndexSpec.builder() diff --git a/server/src/test/java/org/apache/druid/server/coordinator/compact/CompactionStatusTest.java b/server/src/test/java/org/apache/druid/server/coordinator/compact/CompactionStatusTest.java index 0e13f8cd0e1e..49d707581dbd 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/compact/CompactionStatusTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/compact/CompactionStatusTest.java @@ -41,7 +41,7 @@ public void testFindPartitionsSpecWhenGivenIsNull() final ClientCompactionTaskQueryTuningConfig tuningConfig = ClientCompactionTaskQueryTuningConfig.from(null); Assert.assertEquals( - new DynamicPartitionsSpec(null, Long.MAX_VALUE), + new DynamicPartitionsSpec(null, null), CompactionStatus.findPartitionsSpecFromConfig(tuningConfig) ); } @@ -82,52 +82,6 @@ public void testFindPartitionsSpecWhenGivenIsDynamicWithMaxRowsPerSegment() ); } - @Test - public void testFindPartitionsSpecFromConfigWithDeprecatedMaxRowsPerSegmentAndMaxTotalRowsReturnGivenValues() - { - final DataSourceCompactionConfig config = new DataSourceCompactionConfig( - "datasource", - null, - null, - 100, - null, - new UserCompactionTaskQueryTuningConfig( - null, - null, - null, - 1000L, - null, - null, - null, - null, - null, - null, - null, - null, - null, - null, - null, - null, - null, - null, - null - ), - null, - null, - null, - null, - null, - null, - null - ); - Assert.assertEquals( - new DynamicPartitionsSpec(100, 1000L), - CompactionStatus.findPartitionsSpecFromConfig( - ClientCompactionTaskQueryTuningConfig.from(config) - ) - ); - } - @Test public void testFindPartitionsSpecWhenGivenIsHashed() { @@ -160,7 +114,7 @@ private static DataSourceCompactionConfig createCompactionConfig( { return new DataSourceCompactionConfig( DS_WIKI, - null, null, null, null, createTuningConfig(partitionsSpec), + null, null, null, createTuningConfig(partitionsSpec), null, null, null, null, null, null, null ); } @@ -171,7 +125,7 @@ private static UserCompactionTaskQueryTuningConfig createTuningConfig( { return new UserCompactionTaskQueryTuningConfig( null, - null, null, null, null, partitionsSpec, null, null, null, + null, null, null, partitionsSpec, null, null, null, null, null, null, null, null, null, null, null, null, null ); } diff --git a/server/src/test/java/org/apache/druid/server/coordinator/compact/NewestSegmentFirstPolicyTest.java b/server/src/test/java/org/apache/druid/server/coordinator/compact/NewestSegmentFirstPolicyTest.java index 8f24a4ebb7e4..ff02c10df951 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/compact/NewestSegmentFirstPolicyTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/compact/NewestSegmentFirstPolicyTest.java @@ -29,6 +29,7 @@ import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig; import org.apache.druid.common.config.NullHandling; import org.apache.druid.data.input.impl.DimensionsSpec; +import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; import org.apache.druid.indexer.partitions.PartitionsSpec; import org.apache.druid.jackson.DefaultObjectMapper; import org.apache.druid.java.util.common.DateTimes; @@ -1517,7 +1518,7 @@ public void testIteratorReturnsSegmentsAsCompactionStateChangedWithCompactedStat public void testIteratorDoesNotReturnSegmentWithChangingAppendableIndexSpec() { NullHandling.initializeForTests(); - PartitionsSpec partitionsSpec = CompactionStatus.findPartitionsSpecFromConfig(ClientCompactionTaskQueryTuningConfig.from(null)); + final PartitionsSpec partitionsSpec = new DynamicPartitionsSpec(null, 1000L); final SegmentTimeline timeline = createTimeline( new SegmentGenerateSpec( Intervals.of("2017-10-01T00:00:00/2017-10-02T00:00:00"), @@ -1545,7 +1546,6 @@ public void testIteratorDoesNotReturnSegmentWithChangingAppendableIndexSpec() null, new OnheapIncrementalIndex.Spec(true), null, - 1000L, null, partitionsSpec, IndexSpec.DEFAULT, @@ -1580,7 +1580,6 @@ public void testIteratorDoesNotReturnSegmentWithChangingAppendableIndexSpec() null, new OnheapIncrementalIndex.Spec(false), null, - 1000L, null, partitionsSpec, IndexSpec.DEFAULT, @@ -2015,7 +2014,6 @@ private DataSourceCompactionConfig createCompactionConfig( DATA_SOURCE, 0, inputSegmentSizeBytes, - null, skipOffsetFromLatest, tuningConfig, granularitySpec, diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java index 236cfaf7da54..96de76fd348a 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java @@ -720,14 +720,12 @@ public void testCompactWithoutGranularitySpec() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -778,14 +776,12 @@ public void testCompactWithNotNullIOConfig() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -828,14 +824,12 @@ public void testCompactWithNullIOConfig() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -878,14 +872,12 @@ public void testCompactWithGranularitySpec() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -939,14 +931,12 @@ public void testCompactWithDimensionSpec() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -992,14 +982,12 @@ public void testCompactWithoutDimensionSpec() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1042,14 +1030,12 @@ public void testCompactWithRollupInGranularitySpec() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1153,14 +1139,12 @@ public void testCompactWithGranularitySpecConflictWithActiveCompactionTask() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1274,14 +1258,12 @@ public void testCompactWithTransformSpec() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1325,14 +1307,12 @@ public void testCompactWithoutCustomSpecs() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1378,14 +1358,12 @@ public void testCompactWithMetricsSpec() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1459,14 +1437,12 @@ public void testDetermineSegmentGranularityFromSegmentsToCompact() dataSourceName, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1546,14 +1522,12 @@ public void testDetermineSegmentGranularityFromSegmentGranularityInCompactionCon dataSourceName, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1604,14 +1578,12 @@ public void testCompactWithMetricsSpecShouldSetPreserveExistingMetricsTrue() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1657,14 +1629,12 @@ public void testCompactWithoutMetricsSpecShouldSetPreserveExistingMetricsFalse() dataSource, 0, 500L, - null, new Period("PT0H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, @@ -1985,14 +1955,12 @@ private List createCompactionConfigs( dataSource, 0, 50L, - null, new Period("PT1H"), // smaller than segment interval new UserCompactionTaskQueryTuningConfig( null, null, null, null, - null, partitionsSpec, null, null, diff --git a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java index 3d441d9b06d9..c07e98a9c73e 100644 --- a/server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java +++ b/server/src/test/java/org/apache/druid/server/coordinator/duty/KillCompactionConfigTest.java @@ -150,7 +150,6 @@ public void testRunRemoveInactiveDatasourceCompactionConfig() inactiveDatasourceName, null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), @@ -166,7 +165,6 @@ public void testRunRemoveInactiveDatasourceCompactionConfig() activeDatasourceName, null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), @@ -249,7 +247,6 @@ public void testRunRetryForRetryableException() inactiveDatasourceName, null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), diff --git a/server/src/test/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResourceTest.java b/server/src/test/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResourceTest.java index 17db22854779..77c802b82bed 100644 --- a/server/src/test/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResourceTest.java +++ b/server/src/test/java/org/apache/druid/server/http/CoordinatorCompactionConfigsResourceTest.java @@ -58,7 +58,6 @@ public class CoordinatorCompactionConfigsResourceTest "oldDataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), @@ -73,7 +72,6 @@ public class CoordinatorCompactionConfigsResourceTest "newDataSource", null, 500L, - null, new Period(1800), null, new UserCompactionTaskGranularityConfig(Granularities.DAY, null, null), @@ -187,7 +185,6 @@ public void testAddOrUpdateCompactionConfigWithExistingConfig() "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, true), @@ -230,7 +227,6 @@ public void testDeleteCompactionConfigWithExistingConfig() datasourceName, null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), @@ -389,7 +385,6 @@ public void testAddOrUpdateCompactionConfigWithoutExistingConfig() "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), @@ -445,7 +440,6 @@ public void testAddOrUpdateCompactionConfigWithoutExistingConfigAndEngineAsNull( "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null), @@ -486,7 +480,6 @@ public void testAddOrUpdateCompactionConfigWithInvalidMaxNumTasksForMSQEngine() "dataSource", null, 500L, - null, new Period(3600), null, new UserCompactionTaskGranularityConfig(Granularities.HOUR, null, null),