From 10505d4ff18bc51263f71da71a1560cc71ecebb3 Mon Sep 17 00:00:00 2001
From: Maytas Monsereenusorn <maytasm@apache.org>
Date: Wed, 8 Oct 2025 23:30:13 -0700
Subject: [PATCH] Enhance Compaction task to be able to write to a
 different/new datasource

---
 .idea/inspectionProfiles/Druid.xml            | 313 ++++++++++--------
 .../embedded/compact/CompactionTaskTest.java  |   4 +-
 .../common/task/CompactionInputSpec.java      |   5 +
 .../common/task/CompactionIntervalSpec.java   |  25 +-
 .../indexing/common/task/CompactionTask.java  |  15 +-
 .../common/task/SpecificSegmentsSpec.java     |  21 +-
 .../ClientCompactionTaskQuerySerdeTest.java   |   2 +-
 .../common/task/CompactionInputSpecTest.java  |   6 +-
 .../task/CompactionTaskParallelRunTest.java   |  32 +-
 .../common/task/CompactionTaskRunTest.java    |  16 +-
 .../common/task/CompactionTaskTest.java       |  75 +++--
 .../msq/indexing/MSQCompactionRunnerTest.java |   2 +-
 12 files changed, 299 insertions(+), 217 deletions(-)

diff --git a/.idea/inspectionProfiles/Druid.xml b/.idea/inspectionProfiles/Druid.xml
index 0889de4cc67a..88d0e34a54c7 100644
--- a/.idea/inspectionProfiles/Druid.xml
+++ b/.idea/inspectionProfiles/Druid.xml
@@ -39,9 +39,13 @@
       <option name="SUGGEST_NULLABLE_ANNOTATIONS" value="false" />
       <option name="DONT_REPORT_TRUE_ASSERT_STATEMENTS" value="true" />
     </inspection_tool>
+    <inspection_tool class="ConstantValue" enabled="true" level="WARNING" enabled_by_default="true">
+      <option name="DONT_REPORT_TRUE_ASSERT_STATEMENTS" value="true" />
+    </inspection_tool>
     <inspection_tool class="Contract" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="CopyConstructorMissesField" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="CovariantEquals" enabled="true" level="ERROR" enabled_by_default="true" />
+    <inspection_tool class="DataFlowIssueMerged" />
     <inspection_tool class="DuplicateBooleanBranch" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="DuplicateCondition" enabled="true" level="ERROR" enabled_by_default="true">
       <option name="ignoreSideEffectConditions" value="true" />
@@ -59,6 +63,7 @@
     <inspection_tool class="EqualsUsesNonFinalVariable" enabled="true" level="WARNING" enabled_by_default="true" />
     <inspection_tool class="EqualsWhichDoesntCheckParameterClass" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="EqualsWithItself" enabled="true" level="ERROR" enabled_by_default="true" />
+    <inspection_tool class="Eslint" enabled="true" level="WARNING" enabled_by_default="true" />
     <inspection_tool class="FieldAccessNotGuarded" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="FieldCanBeLocal" enabled="true" level="WARNING" enabled_by_default="true">
       <option name="EXCLUDE_ANNOS">
@@ -70,7 +75,6 @@
       </option>
       <option name="IGNORE_FIELDS_USED_IN_MULTIPLE_METHODS" value="true" />
     </inspection_tool>
-    <inspection_tool class="FieldMayBeFinal" enabled="true" level="WARNING" enabled_by_default="true" />
     <inspection_tool class="FinalStaticMethod" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="FlowJSError" enabled="false" level="Non-TeamCity Error" enabled_by_default="false" />
     <inspection_tool class="ForCanBeForeach" enabled="true" level="WARNING" enabled_by_default="true">
@@ -82,7 +86,6 @@
       <option name="ignoreUntypedCollections" value="false" />
     </inspection_tool>
     <inspection_tool class="ForLoopThatDoesntUseLoopVariable" enabled="true" level="ERROR" enabled_by_default="true" />
-    <inspection_tool class="Guava" enabled="false" level="WARNING" enabled_by_default="false" />
     <inspection_tool class="HashCodeUsesNonFinalVariable" enabled="true" level="WARNING" enabled_by_default="true" />
     <inspection_tool class="ImplicitArrayToString" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="IndexOfReplaceableByContains" enabled="true" level="ERROR" enabled_by_default="true" />
@@ -111,6 +114,24 @@
     <inspection_tool class="LengthOneStringInIndexOf" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="ListIndexOfReplaceableByContains" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="MalformedFormatString" enabled="true" level="ERROR" enabled_by_default="true">
+      <option name="classNames">
+        <list>
+          <option value="org.apache.druid.java.util.common.StringUtils" />
+          <option value="org.apache.druid.java.util.common.logger.Logger" />
+        </list>
+      </option>
+      <option name="methodNames">
+        <list>
+          <option value="trace" />
+          <option value="debug" />
+          <option value="info" />
+          <option value="warn" />
+          <option value="error" />
+          <option value="wtf" />
+          <option value="format" />
+          <option value="nonStrictFormat" />
+        </list>
+      </option>
       <option name="additionalClasses" value="org.apache.druid.java.util.common.StringUtils,org.apache.druid.java.util.common.logger.Logger" />
       <option name="additionalMethods" value="trace,debug,info,warn,error,wtf,format,nonStrictFormat" />
     </inspection_tool>
@@ -154,6 +175,7 @@
     </inspection_tool>
     <inspection_tool class="ObjectEqualsNull" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="ObjectToString" enabled="true" level="ERROR" enabled_by_default="true" />
+    <inspection_tool class="OptionalOfNullableMisuseMerged" />
     <inspection_tool class="OverwrittenKey" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="PointlessIndexOfComparison" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="PrimitiveArrayArgumentToVariableArgMethod" enabled="true" level="ERROR" enabled_by_default="true" />
@@ -177,105 +199,6 @@
       <option name="m_ignorePrivateMethods" value="false" />
     </inspection_tool>
     <inspection_tool class="SSBasedInspection" enabled="true" level="ERROR" enabled_by_default="true">
-      <searchConfiguration name="Suboptimal IndexedInts iteration" text="$x$ &lt; $y$.size()" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="x" within="" contains="" />
-        <constraint name="y" nameOfExprType="IndexedInts" exprTypeWithinHierarchy="true" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Lists.newArrayList() with a single argument. Use Collections.singletonList() instead" created="1532737126203" text="Lists.newArrayList($x$)" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="x" nameOfExprType="java\.lang\.Iterable|java\.util\.Iterator|Object\[\]" exprTypeWithinHierarchy="true" negateName="true" negateExprType="true" within="" contains="" />
-        <constraint name="__context__" target="true" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Math.abs(rnd.nextInt()) doensn't guarantee positive result. Use nextInt() &amp; Integer.MAX_VALUE or nextInt(Integer.MAX_VALUE)" created="1535067616084" text="$Math$.abs($x$.nextInt())" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="x" nameOfFormalType="java\.util\.Random" exprTypeWithinHierarchy="true" formalTypeWithinHierarchy="true" within="" contains="" />
-        <constraint name="Math" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Math.abs(rnd.nextLong()) doesn't guarantee positive result. Use nextLong() &amp; Long.MAX_VALUE" created="1535067616084" text="$Math$.abs($x$.nextLong())" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="x" nameOfFormalType="java\.util\.Random" exprTypeWithinHierarchy="true" formalTypeWithinHierarchy="true" within="" contains="" />
-        <constraint name="Math" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use nextInt(bound) instead" created="1535068047572" text="$x$.nextInt() % $a$" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="x" nameOfFormalType="java\.util\.Random" exprTypeWithinHierarchy="true" formalTypeWithinHierarchy="true" within="" contains="" />
-        <constraint name="a" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use ZKPaths.makePath() with many arguments" created="1537504371779" text="org.apache.curator.utils.ZKPaths.makePath(org.apache.curator.utils.ZKPaths.makePath($x$, $y$), $z$)" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="x" within="" contains="" />
-        <constraint name="y" within="" contains="" />
-        <constraint name="z" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use RE (a Druid's class)" created="1539352150701" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" regexp="java\.lang\.RuntimeException" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use RE (a Druid's class) with cause" created="1539353059868" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$),$y$)" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" regexp="java\.lang\.RuntimeException" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
-        <constraint name="y" minCount="0" maxCount="2147483647" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use ISE (a Druid's class)" created="1539353519594" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" regexp="java\.lang\.IllegalStateException" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use ISE (a Druid's class) with cause" created="1539353595734" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$),$y$)" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" regexp="java\.lang\.IllegalStateException" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
-        <constraint name="y" minCount="0" maxCount="2147483647" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use IAE (a Druid's class)" created="1539353691746" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" regexp="java\.lang\.IllegalArgumentException" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use IAE (a Druid's class) with cause" created="1539353766336" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$),$y$)" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" regexp="java\.lang\.IllegalArgumentException" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
-        <constraint name="y" minCount="0" maxCount="2147483647" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use IOE (a Druid's class)" created="1539353913074" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" regexp="java\.io\.IOException" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use IOE (a Druid's class) with cause" created="1539354009031" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$),$y$)" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" regexp="java\.io\.IOException" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
-        <constraint name="y" minCount="0" maxCount="2147483647" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use UOE (a Druid's class)" created="1539354091201" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" regexp="java\.lang\.UnsupportedOperationException" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use TypeReference&lt;List&lt;...&gt;&gt; instead" created="1539884261626" text="TypeReference&lt;ArrayList&lt;$E$&gt;&gt;" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use TypeReference&lt;Map&lt;...&gt;&gt; instead" created="1539884261626" text="TypeReference&lt;HashMap&lt;$K$, $V$&gt;&gt;" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="K" within="" contains="" />
-        <constraint name="V" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use TypeReference&lt;Set&lt;...&gt;&gt; instead" created="1539884261626" text="TypeReference&lt;HashSet&lt;$E$&gt;&gt;" recursive="false" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="E" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Concurrent maps should be assigned into variables of ConcurrentMap type or more specific" text="Map&lt;$K$, $V$&gt; $x$ = $y$;" recursive="true" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="K" within="" contains="" />
-        <constraint name="V" within="" contains="" />
-        <constraint name="x" within="" contains="" />
-        <constraint name="y" nameOfExprType="java\.util\.concurrent\.ConcurrentMap" exprTypeWithinHierarchy="true" within="" contains="" />
-      </searchConfiguration>
       <searchConfiguration name="A ConcurrentHashMap on which compute() is called should be assinged into variables of ConcurrentHashMap type, not ConcurrentMap" text="$x$.compute($y$, $z$)" recursive="true" caseInsensitive="true" type="JAVA">
         <constraint name="__context__" target="true" within="" contains="" />
         <constraint name="x" nameOfExprType="java\.util\.concurrent\.ConcurrentMap" within="" contains="" />
@@ -301,11 +224,46 @@
         <constraint name="b" within="" contains="" />
         <constraint name="c" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="No need to wrap a collection as unmodifiable before addAll()" text="$c$.addAll(Collections.$unmodifiableMethod$($c2$))" recursive="true" caseInsensitive="true" type="JAVA">
+      <searchConfiguration name="Assign an ExecutorService instance to an ExecutorService variable, not an Executor variable" text="$x$ = $y$;" recursive="true" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" within="" contains="" />
+        <constraint name="x" nameOfExprType="java\.util\.concurrent\.Executor" within="" contains="" />
+        <constraint name="y" nameOfExprType="java\.util\.concurrent\.ExecutorService" exprTypeWithinHierarchy="true" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Concurrent maps should be assigned into variables of ConcurrentMap type or more specific" text="Map&lt;$K$, $V$&gt; $x$ = $y$;" recursive="true" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="K" within="" contains="" />
+        <constraint name="V" within="" contains="" />
+        <constraint name="x" within="" contains="" />
+        <constraint name="y" nameOfExprType="java\.util\.concurrent\.ConcurrentMap" exprTypeWithinHierarchy="true" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Create a simple ExecutorService (not scheduled)" text="$x$ = $y$;" recursive="true" caseInsensitive="true" type="JAVA" pattern_context="default">
+        <constraint name="__context__" within="" contains="" />
+        <constraint name="x" nameOfExprType="java\.util\.concurrent\.ExecutorService" within="" contains="" />
+        <constraint name="y" nameOfExprType="java\.util\.concurrent\.ScheduledExecutorService" exprTypeWithinHierarchy="true" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Intialize an ExecutorService instance to an ExecutorService variable, not an Executor variable" text="Executor $x$ = $y$;" recursive="true" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" within="" contains="" />
+        <constraint name="x" within="" contains="" />
+        <constraint name="y" nameOfExprType="java\.util\.concurrent\.ExecutorService" exprTypeWithinHierarchy="true" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Lists.newArrayList() with a single argument. Use Collections.singletonList() instead" created="1532737126203" text="Lists.newArrayList($x$)" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="x" nameOfExprType="java\.lang\.Iterable|java\.util\.Iterator|Object\[\]" exprTypeWithinHierarchy="true" negateName="true" negateExprType="true" within="" contains="" />
+        <constraint name="__context__" target="true" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Math.abs(rnd.nextInt()) doensn't guarantee positive result. Use nextInt() &amp; Integer.MAX_VALUE or nextInt(Integer.MAX_VALUE)" created="1535067616084" text="$Math$.abs($x$.nextInt())" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="x" nameOfFormalType="java\.util\.Random" exprTypeWithinHierarchy="true" formalTypeWithinHierarchy="true" within="" contains="" />
+        <constraint name="Math" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Math.abs(rnd.nextLong()) doesn't guarantee positive result. Use nextLong() &amp; Long.MAX_VALUE" created="1535067616084" text="$Math$.abs($x$.nextLong())" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="x" nameOfFormalType="java\.util\.Random" exprTypeWithinHierarchy="true" formalTypeWithinHierarchy="true" within="" contains="" />
+        <constraint name="Math" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="new Collection(Arrays.asList()), or some utility from CollectionUtils (or create an utility)" text="Stream.of($x$).collect($c$())" recursive="true" caseInsensitive="true" type="JAVA">
         <constraint name="__context__" target="true" within="" contains="" />
         <constraint name="c" within="" contains="" />
-        <constraint name="unmodifiableMethod" regexp="unmodifiable.*" within="" contains="" />
-        <constraint name="c2" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
       </searchConfiguration>
       <searchConfiguration name="No need to copy a collection as immutable before addAll()" text="$m$.addAll($ImmutableCollection$.copyOf($x$))" recursive="true" caseInsensitive="true" type="JAVA">
         <constraint name="__context__" target="true" within="" contains="" />
@@ -313,22 +271,23 @@
         <constraint name="x" maxCount="2147483647" within="" contains="" />
         <constraint name="ImmutableCollection" regexp="Immutable.*" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="No need to wrap a map as unmodifiable before putAll()" text="$m$.putAll(Collections.$unmodifiableMapMethod$($m2$))" recursive="true" caseInsensitive="true" type="JAVA">
-        <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="m" within="" contains="" />
-        <constraint name="m2" within="" contains="" />
-        <constraint name="unmodifiableMapMethod" regexp="unmodifiable.*" within="" contains="" />
-      </searchConfiguration>
       <searchConfiguration name="No need to copy a map to immutable before putAll()" text="$m$.putAll($ImmutableMap$.copyOf($x$))" recursive="true" caseInsensitive="true" type="JAVA">
         <constraint name="__context__" target="true" within="" contains="" />
         <constraint name="m" within="" contains="" />
         <constraint name="x" maxCount="2147483647" within="" contains="" />
         <constraint name="ImmutableMap" regexp="Immutable.*" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="Use map.putIfAbsent(k,v) or map.computeIfAbsent(k,v) where appropriate instead of containsKey() + put(). If computing v is expensive or has side effects use map.computeIfAbsent() instead" created="1558868694225" text="if (!$m$.containsKey($k$)) {&#10;    $m$.put($k$, $v$);&#10;}" recursive="false" caseInsensitive="true" type="JAVA">
+      <searchConfiguration name="No need to wrap a collection as unmodifiable before addAll()" text="$c$.addAll(Collections.$unmodifiableMethod$($c2$))" recursive="true" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="c" within="" contains="" />
+        <constraint name="unmodifiableMethod" regexp="unmodifiable.*" within="" contains="" />
+        <constraint name="c2" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="No need to wrap a map as unmodifiable before putAll()" text="$m$.putAll(Collections.$unmodifiableMapMethod$($m2$))" recursive="true" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
         <constraint name="m" within="" contains="" />
-        <constraint name="k" within="" contains="" />
-        <constraint name="v" within="" contains="" />
+        <constraint name="m2" within="" contains="" />
+        <constraint name="unmodifiableMapMethod" regexp="unmodifiable.*" within="" contains="" />
       </searchConfiguration>
       <searchConfiguration name="Prohibit Thread.getState() != TERMINATED antipattern" text="$t$.getState()!=Thread.State.$state$" recursive="true" caseInsensitive="true" type="JAVA">
         <constraint name="t" within="" contains="" />
@@ -338,16 +297,10 @@
         <constraint name="t" within="" contains="" />
         <constraint name="state" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="Use CollectionUtils.mapValues(Map&lt;K,V&gt;, Function&lt;V,V2&gt;)" text="$x$.entrySet().stream().collect(Collectors.toMap($k$ -&gt; $k$.getKey(), $y$))" recursive="true" caseInsensitive="true" type="JAVA">
-        <constraint name="x" within="" contains="" />
-        <constraint name="y" within="" contains="" />
-        <constraint name="k" within="" contains="" />
+      <searchConfiguration name="Suboptimal IndexedInts iteration" text="$x$ &lt; $y$.size()" recursive="false" caseInsensitive="true" type="JAVA">
         <constraint name="__context__" target="true" within="" contains="" />
-      </searchConfiguration>
-      <searchConfiguration name="Use CollectionUtils.mapValues(Map&lt;k,v&gt;, Function&lt;v,v2&gt;)" text="$x$.entrySet().stream().collect(Collectors.toMap(Entry::getKey, $y$))" recursive="true" caseInsensitive="true" type="JAVA">
         <constraint name="x" within="" contains="" />
-        <constraint name="y" within="" contains="" />
-        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="y" nameOfExprType="IndexedInts" exprTypeWithinHierarchy="true" within="" contains="" />
       </searchConfiguration>
       <searchConfiguration name="Use CollectionUtils.mapKeys(Map&lt;K,V&gt;, Function&lt;K,K2&gt;)" text="$x$.entrySet().stream().collect(Collectors.toMap($y$, $v$ -&gt; $v$.getValue()))" recursive="true" caseInsensitive="true" type="JAVA">
         <constraint name="x" within="" contains="" />
@@ -360,10 +313,16 @@
         <constraint name="y" within="" contains="" />
         <constraint name="__context__" target="true" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="new Collection(Arrays.asList()), or some utility from CollectionUtils (or create an utility)" text="Stream.of($x$).collect($c$())" recursive="true" caseInsensitive="true" type="JAVA">
+      <searchConfiguration name="Use CollectionUtils.mapValues(Map&lt;K,V&gt;, Function&lt;V,V2&gt;)" text="$x$.entrySet().stream().collect(Collectors.toMap($k$ -&gt; $k$.getKey(), $y$))" recursive="true" caseInsensitive="true" type="JAVA">
+        <constraint name="x" within="" contains="" />
+        <constraint name="y" within="" contains="" />
+        <constraint name="k" within="" contains="" />
+        <constraint name="__context__" target="true" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use CollectionUtils.mapValues(Map&lt;k,v&gt;, Function&lt;v,v2&gt;)" text="$x$.entrySet().stream().collect(Collectors.toMap(Entry::getKey, $y$))" recursive="true" caseInsensitive="true" type="JAVA">
+        <constraint name="x" within="" contains="" />
+        <constraint name="y" within="" contains="" />
         <constraint name="__context__" target="true" within="" contains="" />
-        <constraint name="c" within="" contains="" />
-        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
       </searchConfiguration>
       <searchConfiguration name="Use computeIfAbsent(...).somethingElse(...) chain instead" text="$m$.putIfAbsent($k$, $l$);&#10;$m$.get($k$).$x$($y$);" recursive="true" caseInsensitive="true" type="JAVA">
         <constraint name="__context__" target="true" within="" contains="" />
@@ -373,40 +332,103 @@
         <constraint name="l" within="" contains="" />
         <constraint name="y" minCount="0" maxCount="2147483647" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="Assign an ExecutorService instance to an ExecutorService variable, not an Executor variable" text="$x$ = $y$;" recursive="true" caseInsensitive="true" type="JAVA" >
+      <searchConfiguration name="Use equals on Pattern.toString()" text="$a$.equals($b$)" recursive="true" caseInsensitive="true" type="JAVA" pattern_context="default">
         <constraint name="__context__" within="" contains="" />
-        <constraint name="x" nameOfExprType="java\.util\.concurrent\.Executor" within="" contains="" />
-        <constraint name="y" nameOfExprType="java\.util\.concurrent\.ExecutorService" exprTypeWithinHierarchy="true" within="" contains="" />
+        <constraint name="a" nameOfExprType="Pattern" within="" contains="" />
+        <constraint name="b" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="Intialize an ExecutorService instance to an ExecutorService variable, not an Executor variable" text="Executor $x$ = $y$;" recursive="true" caseInsensitive="true" type="JAVA" >
+      <searchConfiguration name="Use hashCode of Pattern.toString() instead" text="$a$.hashCode()" recursive="true" caseInsensitive="true" type="JAVA" pattern_context="default">
         <constraint name="__context__" within="" contains="" />
-        <constraint name="x"  within="" contains="" />
-        <constraint name="y" nameOfExprType="java\.util\.concurrent\.ExecutorService" exprTypeWithinHierarchy="true"  within="" contains="" />
+        <constraint name="a" nameOfExprType="java\.util\.regex\.Pattern" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="Create a simple ExecutorService (not scheduled)" text="$x$ = $y$;" recursive="true" caseInsensitive="true" type="JAVA" pattern_context="default">
-        <constraint name="__context__" within="" contains="" />
-        <constraint name="x" nameOfExprType="java\.util\.concurrent\.ExecutorService" within="" contains="" />
-        <constraint name="y" nameOfExprType="java\.util\.concurrent\.ScheduledExecutorService" exprTypeWithinHierarchy="true" within="" contains="" />
+      <searchConfiguration name="Use IAE (a Druid's class)" created="1539353691746" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" regexp="java\.lang\.IllegalArgumentException" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="Use equals on Pattern.toString()" text="$a$.equals($b$)" recursive="true" caseInsensitive="true" type="JAVA" pattern_context="default">
-        <constraint name="__context__" within="" contains="" />
-        <constraint name="a" nameOfExprType="Pattern" within="" contains="" />
-        <constraint name="b" within="" contains="" />
+      <searchConfiguration name="Use IAE (a Druid's class) with cause" created="1539353766336" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$),$y$)" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" regexp="java\.lang\.IllegalArgumentException" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
+        <constraint name="y" minCount="0" maxCount="2147483647" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use IOE (a Druid's class)" created="1539353913074" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" regexp="java\.io\.IOException" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use IOE (a Druid's class) with cause" created="1539354009031" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$),$y$)" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" regexp="java\.io\.IOException" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
+        <constraint name="y" minCount="0" maxCount="2147483647" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use ISE (a Druid's class)" created="1539353519594" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" regexp="java\.lang\.IllegalStateException" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use ISE (a Druid's class) with cause" created="1539353595734" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$),$y$)" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" regexp="java\.lang\.IllegalStateException" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
+        <constraint name="y" minCount="0" maxCount="2147483647" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use map.putIfAbsent(k,v) or map.computeIfAbsent(k,v) where appropriate instead of containsKey() + put(). If computing v is expensive or has side effects use map.computeIfAbsent() instead" created="1558868694225" text="if (!$m$.containsKey($k$)) {&#10;    $m$.put($k$, $v$);&#10;}" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="m" within="" contains="" />
+        <constraint name="k" within="" contains="" />
+        <constraint name="v" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use nextInt(bound) instead" created="1535068047572" text="$x$.nextInt() % $a$" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="x" nameOfFormalType="java\.util\.Random" exprTypeWithinHierarchy="true" formalTypeWithinHierarchy="true" within="" contains="" />
+        <constraint name="a" within="" contains="" />
       </searchConfiguration>
       <searchConfiguration name="Use Objects.equals() on Pattern.toString()" text="java.util.Objects.equals($a$, $b$)" recursive="true" caseInsensitive="true" type="JAVA" pattern_context="default">
         <constraint name="__context__" within="" contains="" />
         <constraint name="a" nameOfExprType="Pattern" within="" contains="" />
         <constraint name="b" within="" contains="" />
       </searchConfiguration>
-      <searchConfiguration name="Use hashCode of Pattern.toString() instead" text="$a$.hashCode()" recursive="true" caseInsensitive="true" type="JAVA" pattern_context="default">
-        <constraint name="__context__" within="" contains="" />
-        <constraint name="a" nameOfExprType="java\.util\.regex\.Pattern" within="" contains="" />
-      </searchConfiguration>
       <searchConfiguration name="Use Objects.hash() on Pattern.toString() instead" text="java.util.Objects.hash($b$,$a$)" recursive="true" caseInsensitive="true" type="JAVA" pattern_context="default">
         <constraint name="__context__" within="" contains="" />
         <constraint name="a" nameOfExprType="java\.util\.regex\.Pattern" within="" contains="" />
         <constraint name="b" minCount="0" maxCount="2147483647" within="" contains="" />
       </searchConfiguration>
+      <searchConfiguration name="Use RE (a Druid's class)" created="1539352150701" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" regexp="java\.lang\.RuntimeException" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use RE (a Druid's class) with cause" created="1539353059868" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$),$y$)" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" regexp="java\.lang\.RuntimeException" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
+        <constraint name="y" minCount="0" maxCount="2147483647" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use TypeReference&lt;List&lt;...&gt;&gt; instead" created="1539884261626" text="TypeReference&lt;ArrayList&lt;$E$&gt;&gt;" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use TypeReference&lt;Map&lt;...&gt;&gt; instead" created="1539884261626" text="TypeReference&lt;HashMap&lt;$K$, $V$&gt;&gt;" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="K" within="" contains="" />
+        <constraint name="V" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use TypeReference&lt;Set&lt;...&gt;&gt; instead" created="1539884261626" text="TypeReference&lt;HashSet&lt;$E$&gt;&gt;" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use UOE (a Druid's class)" created="1539354091201" text="new $E$(org.apache.druid.java.util.common.StringUtils.format($x$))" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="E" regexp="java\.lang\.UnsupportedOperationException" within="" contains="" />
+        <constraint name="x" minCount="0" maxCount="2147483647" within="" contains="" />
+      </searchConfiguration>
+      <searchConfiguration name="Use ZKPaths.makePath() with many arguments" created="1537504371779" text="org.apache.curator.utils.ZKPaths.makePath(org.apache.curator.utils.ZKPaths.makePath($x$, $y$), $z$)" recursive="false" caseInsensitive="true" type="JAVA">
+        <constraint name="__context__" target="true" within="" contains="" />
+        <constraint name="x" within="" contains="" />
+        <constraint name="y" within="" contains="" />
+        <constraint name="z" within="" contains="" />
+      </searchConfiguration>
     </inspection_tool>
     <inspection_tool class="SimplifyStreamApiCallChains" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="SpellCheckingInspection" enabled="false" level="TYPO" enabled_by_default="false">
@@ -427,6 +449,7 @@
     <inspection_tool class="StringEquality" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="StringEqualsCharSequence" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="StringTokenizerDelimiter" enabled="true" level="ERROR" enabled_by_default="true" />
+    <inspection_tool class="Stylelint" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="SubtractionInCompareTo" enabled="true" level="ERROR" enabled_by_default="true" />
     <inspection_tool class="SuspiciousArrayCast" enabled="true" level="WARNING" enabled_by_default="true" />
     <inspection_tool class="SuspiciousArrayMethodCall" enabled="true" level="ERROR" enabled_by_default="true" />
@@ -481,8 +504,8 @@
     <inspection_tool class="XmlHighlighting" enabled="true" level="Non-TeamCity Warning" enabled_by_default="true" />
     <inspection_tool class="XmlInvalidId" enabled="true" level="Non-TeamCity Error" enabled_by_default="true" />
     <inspection_tool class="XmlPathReference" enabled="true" level="Non-TeamCity Error" enabled_by_default="true" />
-    <inspection_tool class="unused" enabled="true" level="WARNING" enabled_by_default="true" isSelected="false">
-      <scope name="UnusedInspectionsScope" level="ERROR" enabled="true" isSelected="false">
+    <inspection_tool class="unused" enabled="true" level="WARNING" enabled_by_default="true" checkParameterExcludingHierarchy="false">
+      <scope name="UnusedInspectionsScope" level="ERROR" enabled="true" checkParameterExcludingHierarchy="false">
         <option name="LOCAL_VARIABLE" value="true" />
         <option name="FIELD" value="true" />
         <option name="METHOD" value="true" />
@@ -493,6 +516,7 @@
         <option name="ADD_APPLET_TO_ENTRIES" value="true" />
         <option name="ADD_SERVLET_TO_ENTRIES" value="true" />
         <option name="ADD_NONJAVA_TO_ENTRIES" value="true" />
+        <option name="selected" value="true" />
       </scope>
       <option name="LOCAL_VARIABLE" value="true" />
       <option name="FIELD" value="true" />
@@ -504,6 +528,7 @@
       <option name="ADD_APPLET_TO_ENTRIES" value="true" />
       <option name="ADD_SERVLET_TO_ENTRIES" value="true" />
       <option name="ADD_NONJAVA_TO_ENTRIES" value="true" />
+      <option name="selected" value="true" />
     </inspection_tool>
   </profile>
 </component>
\ No newline at end of file
diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTaskTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTaskTest.java
index 84ee947c8467..ab9be6408032 100644
--- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTaskTest.java
+++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/CompactionTaskTest.java
@@ -87,7 +87,7 @@ public class CompactionTaskTest extends CompactionTestBase
       () -> TaskBuilder
           .ofTypeCompact()
           .context("storeCompactionState", true)
-          .ioConfig(new CompactionIntervalSpec(Intervals.of("2013-08-31/2013-09-02"), null), false);
+          .ioConfig(new CompactionIntervalSpec(Intervals.of("2013-08-31/2013-09-02"), null, null), false);
   private static final Supplier<TaskBuilder.Compact> PARALLEL_COMPACTION_TASK =
       () -> COMPACTION_TASK.get().tuningConfig(
           t -> t.withPartitionsSpec(new HashedPartitionsSpec(null, null, null))
@@ -98,7 +98,7 @@ public class CompactionTaskTest extends CompactionTestBase
       () -> TaskBuilder
           .ofTypeCompact()
           .context("storeCompactionState", true)
-          .ioConfig(new CompactionIntervalSpec(Intervals.of("2013-08-31/2013-09-02"), null), true);
+          .ioConfig(new CompactionIntervalSpec(Intervals.of("2013-08-31/2013-09-02"), null, null), true);
 
   private static final Supplier<TaskBuilder.Index> INDEX_TASK_WITH_TIMESTAMP =
       () -> MoreResources.Task.INDEX_TASK_WITH_AGGREGATORS.get().dimensions(
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionInputSpec.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionInputSpec.java
index d73a0ffe39a0..50b5f06a1cc0 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionInputSpec.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionInputSpec.java
@@ -51,4 +51,9 @@ public interface CompactionInputSpec
    * @param latestSegments       most recent published segments in the interval returned by {@link #findInterval}
    */
   boolean validateSegments(LockGranularity lockGranularityInUse, List<DataSegment> latestSegments);
+
+  /**
+   * Return the datasource to be used as input to the compaction task.
+   */
+  String getDataSource();
 }
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionIntervalSpec.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionIntervalSpec.java
index 40a4d775e031..1493e08facea 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionIntervalSpec.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionIntervalSpec.java
@@ -43,11 +43,14 @@ public class CompactionIntervalSpec implements CompactionInputSpec
   private final Interval interval;
   @Nullable
   private final String sha256OfSortedSegmentIds;
+  @Nullable
+  private final String dataSource;
 
   @JsonCreator
   public CompactionIntervalSpec(
       @JsonProperty("interval") Interval interval,
-      @JsonProperty("sha256OfSortedSegmentIds") @Nullable String sha256OfSortedSegmentIds
+      @JsonProperty("sha256OfSortedSegmentIds") @Nullable String sha256OfSortedSegmentIds,
+      @JsonProperty("dataSource") @Nullable String dataSource
   )
   {
     if (interval != null && interval.toDurationMillis() == 0) {
@@ -55,6 +58,7 @@ public CompactionIntervalSpec(
     }
     this.interval = interval;
     this.sha256OfSortedSegmentIds = sha256OfSortedSegmentIds;
+    this.dataSource = dataSource;
   }
 
   @JsonProperty
@@ -70,6 +74,14 @@ public String getSha256OfSortedSegmentIds()
     return sha256OfSortedSegmentIds;
   }
 
+  @Override
+  @Nullable
+  @JsonProperty
+  public String getDataSource()
+  {
+    return dataSource;
+  }
+
   @Override
   public Interval findInterval(String dataSource)
   {
@@ -97,21 +109,19 @@ public boolean validateSegments(LockGranularity lockGranularityInUse, List<DataS
   @Override
   public boolean equals(Object o)
   {
-    if (this == o) {
-      return true;
-    }
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
     CompactionIntervalSpec that = (CompactionIntervalSpec) o;
     return Objects.equals(interval, that.interval) &&
-           Objects.equals(sha256OfSortedSegmentIds, that.sha256OfSortedSegmentIds);
+           Objects.equals(sha256OfSortedSegmentIds, that.sha256OfSortedSegmentIds) &&
+           Objects.equals(dataSource, that.dataSource);
   }
 
   @Override
   public int hashCode()
   {
-    return Objects.hash(interval, sha256OfSortedSegmentIds);
+    return Objects.hash(interval, sha256OfSortedSegmentIds, dataSource);
   }
 
   @Override
@@ -119,7 +129,8 @@ public String toString()
   {
     return "CompactionIntervalSpec{" +
            "interval=" + interval +
-           ", sha256OfSegmentIds='" + sha256OfSortedSegmentIds + '\'' +
+           ", sha256OfSortedSegmentIds='" + sha256OfSortedSegmentIds + '\'' +
+           ", dataSource='" + dataSource + '\'' +
            '}';
   }
 }
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 1c65264f6b6d..820a2c9b8955 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
@@ -212,7 +212,7 @@ public CompactionTask(
     if (ioConfig != null) {
       this.ioConfig = ioConfig;
     } else if (interval != null) {
-      this.ioConfig = new CompactionIOConfig(new CompactionIntervalSpec(interval, null), false, null);
+      this.ioConfig = new CompactionIOConfig(new CompactionIntervalSpec(interval, null, null), false, null);
     } else {
       // We already checked segments is not null or empty above.
       //noinspection ConstantConditions
@@ -242,7 +242,10 @@ public CompactionTask(
     }
     this.projections = projections;
     this.tuningConfig = tuningConfig != null ? getTuningConfig(tuningConfig) : null;
-    this.segmentProvider = new SegmentProvider(dataSource, this.ioConfig.getInputSpec());
+    this.segmentProvider = new SegmentProvider(
+        this.ioConfig.getInputSpec().getDataSource() == null ? dataSource : this.ioConfig.getInputSpec().getDataSource(),
+        this.ioConfig.getInputSpec()
+    );
     // Note: The default compactionRunnerType used here should match the default runner used in CompactSegments#run
     // when no runner is detected in the returned compactionTaskQuery.
     this.compactionRunner = compactionRunner == null
@@ -517,6 +520,7 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
     emitMetric(toolbox.getEmitter(), "ingest/count", 1);
 
     final Map<Interval, DataSchema> intervalDataSchemas = createDataSchemasForIntervals(
+        getDataSource(),
         toolbox,
         getTaskLockHelper().getLockGranularityToUse(),
         segmentProvider,
@@ -548,6 +552,7 @@ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
    */
   @VisibleForTesting
   static Map<Interval, DataSchema> createDataSchemasForIntervals(
+      final String dataSource,
       final TaskToolbox toolbox,
       final LockGranularity lockGranularityInUse,
       final SegmentProvider segmentProvider,
@@ -613,7 +618,7 @@ static Map<Interval, DataSchema> createDataSchemasForIntervals(
         final DataSchema dataSchema = createDataSchema(
             toolbox.getEmitter(),
             metricBuilder,
-            segmentProvider.dataSource,
+            dataSource,
             interval,
             lazyFetchSegments(segmentsToCompact, toolbox.getSegmentCacheManager()),
             dimensionsSpec,
@@ -633,7 +638,7 @@ static Map<Interval, DataSchema> createDataSchemasForIntervals(
       final DataSchema dataSchema = createDataSchema(
           toolbox.getEmitter(),
           metricBuilder,
-          segmentProvider.dataSource,
+          dataSource,
           JodaUtils.umbrellaInterval(
               Iterables.transform(
                   timelineSegments,
@@ -1275,7 +1280,7 @@ public Builder(
 
     public Builder interval(Interval interval)
     {
-      return inputSpec(new CompactionIntervalSpec(interval, null));
+      return inputSpec(new CompactionIntervalSpec(interval, null, null));
     }
 
     public Builder segments(List<DataSegment> segments)
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/SpecificSegmentsSpec.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/SpecificSegmentsSpec.java
index 7b26c22a4a54..b5b77e9a8f6d 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/SpecificSegmentsSpec.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/SpecificSegmentsSpec.java
@@ -28,6 +28,7 @@
 import org.apache.druid.timeline.SegmentId;
 import org.joda.time.Interval;
 
+import javax.annotation.Nullable;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
@@ -39,18 +40,26 @@ public class SpecificSegmentsSpec implements CompactionInputSpec
 
   private final List<String> segments;
 
+  @Nullable
+  private final String dataSource;
+
   public static SpecificSegmentsSpec fromSegments(List<DataSegment> segments)
   {
     Preconditions.checkArgument(!segments.isEmpty(), "Empty segment list");
     return new SpecificSegmentsSpec(
-        segments.stream().map(segment -> segment.getId().toString()).collect(Collectors.toList())
+        segments.stream().map(segment -> segment.getId().toString()).collect(Collectors.toList()),
+        null
     );
   }
 
   @JsonCreator
-  public SpecificSegmentsSpec(@JsonProperty("segments") List<String> segments)
+  public SpecificSegmentsSpec(
+      @JsonProperty("segments") List<String> segments,
+      @JsonProperty("dataSource") @Nullable String dataSource
+  )
   {
     this.segments = segments;
+    this.dataSource = dataSource;
     // Sort segments to use in validateSegments.
     Collections.sort(this.segments);
   }
@@ -61,6 +70,14 @@ public List<String> getSegments()
     return segments;
   }
 
+  @Override
+  @Nullable
+  @JsonProperty
+  public String getDataSource()
+  {
+    return dataSource;
+  }
+
   @Override
   public Interval findInterval(String dataSource)
   {
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 1c04c7b5bd2b..ecbee4ce9b96 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
@@ -341,7 +341,7 @@ private CompactionTask createCompactionTask(CompactionTransformSpec transformSpe
         "datasource",
         new SegmentCacheManagerFactory(TestIndex.INDEX_IO, MAPPER)
     )
-        .inputSpec(new CompactionIntervalSpec(Intervals.of("2019/2020"), "testSha256OfSortedSegmentIds"), true)
+        .inputSpec(new CompactionIntervalSpec(Intervals.of("2019/2020"), "testSha256OfSortedSegmentIds", null), true)
         .tuningConfig(
             TuningConfigBuilder
                 .forParallelIndexTask()
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionInputSpecTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionInputSpecTest.java
index 8c0db006bbfb..8814cf9af087 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionInputSpecTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionInputSpecTest.java
@@ -53,12 +53,14 @@ public static Iterable<Object[]> constructorFeeder()
         new Object[]{
             new CompactionIntervalSpec(
                 INTERVAL,
-                SegmentUtils.hashIds(SEGMENTS)
+                SegmentUtils.hashIds(SEGMENTS),
+                null
             )
         },
         new Object[]{
             new SpecificSegmentsSpec(
-                SEGMENTS.stream().map(segment -> segment.getId().toString()).collect(Collectors.toList())
+                SEGMENTS.stream().map(segment -> segment.getId().toString()).collect(Collectors.toList()),
+                null
             )
         }
     );
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskParallelRunTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskParallelRunTest.java
index 53a7259f5bc6..cb3d45b88e72 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskParallelRunTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/CompactionTaskParallelRunTest.java
@@ -178,7 +178,7 @@ public void testRunParallelWithDynamicPartitioningMatchCompactionState()
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .build();
 
@@ -230,7 +230,7 @@ public void testRunParallelWithHashPartitioningMatchCompactionState() throws Exc
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(newTuningConfig(new HashedPartitionsSpec(null, 3, null), 2, true))
         .build();
 
@@ -293,7 +293,7 @@ public void testRunParallelWithRangePartitioning() throws Exception
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(newTuningConfig(new SingleDimensionPartitionsSpec(7, null, "dim", false), 2, true))
         .build();
 
@@ -342,7 +342,7 @@ public void testRunParallelWithRangePartitioningAndNoUpfrontSegmentFetching() th
     );
 
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(newTuningConfig(new SingleDimensionPartitionsSpec(7, null, "dim", false), 2, true))
         .dimensionsSpec(new DimensionsSpec(DimensionsSpec.getDefaultSchemas(ImmutableList.of("ts", "dim"))))
         .metricsSpec(new AggregatorFactory[]{new LongSumAggregatorFactory("val", "val")})
@@ -396,7 +396,7 @@ public void testRunParallelWithMultiDimensionRangePartitioning() throws Exceptio
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(newTuningConfig(
             new DimensionRangePartitionsSpec(7, null, Arrays.asList("dim1", "dim2"), false),
             2,
@@ -447,7 +447,7 @@ public void testRunParallelWithRangePartitioningWithSingleTask() throws Exceptio
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(newTuningConfig(new SingleDimensionPartitionsSpec(7, null, "dim", false), 1, true))
         .build();
 
@@ -495,7 +495,7 @@ public void testRunParallelWithMultiDimensionRangePartitioningWithSingleTask() t
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(newTuningConfig(
             new DimensionRangePartitionsSpec(7, null, Arrays.asList("dim1", "dim2"), false),
             1,
@@ -543,7 +543,7 @@ public void testRunCompactionStateNotStoreIfContextSetToFalse()
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .context(ImmutableMap.of(Tasks.STORE_COMPACTION_STATE_KEY, false))
         .build();
@@ -573,7 +573,7 @@ public void testRunCompactionWithFilterShouldStoreInState() throws Exception
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .transformSpec(new CompactionTransformSpec(new SelectorDimFilter("dim", "a", null)))
         .build();
@@ -624,7 +624,7 @@ public void testRunCompactionWithNewMetricsShouldStoreInState() throws Exception
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .metricsSpec(new AggregatorFactory[]{
             new CountAggregatorFactory("cnt"),
@@ -679,7 +679,7 @@ public void testCompactHashAndDynamicPartitionedSegments()
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .build();
 
@@ -728,7 +728,7 @@ public void testCompactRangeAndDynamicPartitionedSegments()
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .build();
 
@@ -820,7 +820,7 @@ public void testCompactionDropSegmentsOfInputIntervalIfDropFlagIsSet() throws Ex
     );
     final CompactionTask compactionTask = builder
         // Set the dropExisting flag to true in the IOConfig of the compaction task
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null), true)
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null), true)
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.MINUTE, null, null))
         .build();
@@ -865,7 +865,7 @@ public void testCompactionDoesNotDropSegmentsIfDropFlagNotSet() throws Exception
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.MINUTE, null, null))
         .build();
@@ -906,7 +906,7 @@ public void testRunParallelWithProjections()
         getSegmentCacheManagerFactory()
     );
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .build();
 
@@ -959,7 +959,7 @@ public void testRunParallelAddProjections()
                                .aggregators(new LongSumAggregatorFactory("val", "val"))
                                .build();
     final CompactionTask compactionTask = builder
-        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null))
+        .inputSpec(new CompactionIntervalSpec(INTERVAL_TO_INDEX, null, null))
         .tuningConfig(AbstractParallelIndexSupervisorTaskTest.DEFAULT_TUNING_CONFIG_FOR_PARALLEL_INDEXING)
         .projections(
             ImmutableList.of(
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 7e5a15dad079..9edbf1ac06d0 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
@@ -679,7 +679,7 @@ public void testWithSegmentGranularityMisalignedInterval() throws Exception
     final CompactionTask compactionTask1 = compactionTaskBuilder()
         .ioConfig(
             new CompactionIOConfig(
-                new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null),
+                new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null, null),
                 false,
                 null
             )
@@ -709,7 +709,7 @@ public void testWithSegmentGranularityMisalignedIntervalAllowed() throws Excepti
     final CompactionTask compactionTask1 = compactionTaskBuilder()
         .ioConfig(
             new CompactionIOConfig(
-                new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null),
+                new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null, null),
                 true,
                 null
             )
@@ -1111,7 +1111,7 @@ public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullInterva
     final CompactionTask partialCompactionTask = compactionTaskBuilder()
         .segmentGranularity(Granularities.MINUTE)
         // Set dropExisting to true
-        .inputSpec(new CompactionIntervalSpec(compactionPartialInterval, null), true)
+        .inputSpec(new CompactionIntervalSpec(compactionPartialInterval, null, null), true)
         .build();
     final Pair<TaskStatus, DataSegmentsWithSchemas> partialCompactionResult = runTask(partialCompactionTask);
     verifySchema(partialCompactionResult.rhs);
@@ -1173,7 +1173,7 @@ public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullInterva
     final CompactionTask fullCompactionTask = compactionTaskBuilder()
         .segmentGranularity(null)
         // Set dropExisting to true
-        .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null), true)
+        .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null, null), true)
         .build();
 
     // **** FULL COMPACTION ****
@@ -1263,7 +1263,7 @@ public void testCompactDatasourceOverIntervalWithOnlyTombstones() throws Excepti
     final CompactionTask partialCompactionTask = compactionTaskBuilder()
         .segmentGranularity(Granularities.MINUTE)
         // Set dropExisting to true
-        .inputSpec(new CompactionIntervalSpec(compactionPartialInterval, null), true)
+        .inputSpec(new CompactionIntervalSpec(compactionPartialInterval, null, null), true)
         .build();
     final Pair<TaskStatus, DataSegmentsWithSchemas> partialCompactionResult = runTask(partialCompactionTask);
     verifySchema(partialCompactionResult.rhs);
@@ -1317,7 +1317,7 @@ public void testCompactDatasourceOverIntervalWithOnlyTombstones() throws Excepti
         .segmentGranularity(null)
         // Set dropExisting to true
         // last 59 minutes of our 01, should be all tombstones
-        .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01T01:01:00/2014-01-01T02:00:00"), null), true)
+        .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01T01:01:00/2014-01-01T02:00:00"), null, null), true)
         .build();
 
     // **** Compaction over tombstones ****
@@ -1355,7 +1355,7 @@ public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullInterva
     final CompactionTask partialCompactionTask = compactionTaskBuilder()
         .segmentGranularity(Granularities.MINUTE)
         // Set dropExisting to false
-        .inputSpec(new CompactionIntervalSpec(partialInterval, null), false)
+        .inputSpec(new CompactionIntervalSpec(partialInterval, null, null), false)
         .build();
 
     final Pair<TaskStatus, DataSegmentsWithSchemas> partialCompactionResult = runTask(partialCompactionTask);
@@ -1377,7 +1377,7 @@ public void testPartialIntervalCompactWithFinerSegmentGranularityThenFullInterva
     final CompactionTask fullCompactionTask = compactionTaskBuilder()
         .segmentGranularity(null)
         // Set dropExisting to false
-        .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null), false)
+        .inputSpec(new CompactionIntervalSpec(Intervals.of("2014-01-01/2014-01-02"), null, null), false)
         .build();
 
     final Pair<TaskStatus, DataSegmentsWithSchemas> fullCompactionResult = runTask(fullCompactionTask);
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 5d319c1013fd..f7df2d4846a4 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
@@ -375,7 +375,7 @@ public void testCreateCompactionTaskWithGranularitySpec()
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.tuningConfig(createTuningConfig());
     builder.segmentGranularity(Granularities.HOUR);
     final CompactionTask taskCreatedWithSegmentGranularity = builder.build();
@@ -384,7 +384,7 @@ public void testCreateCompactionTaskWithGranularitySpec()
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder2.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder2.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder2.tuningConfig(createTuningConfig());
     builder2.granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.HOUR, Granularities.DAY, null));
     final CompactionTask taskCreatedWithGranularitySpec = builder2.build();
@@ -401,7 +401,7 @@ public void testCreateCompactionTaskWithConflictingGranularitySpecAndSegmentGran
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.tuningConfig(createTuningConfig());
     builder.segmentGranularity(Granularities.HOUR);
     builder.granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.MINUTE, Granularities.DAY, null));
@@ -431,7 +431,7 @@ public void testCreateCompactionTaskWithTransformSpec()
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.tuningConfig(createTuningConfig());
     builder.transformSpec(transformSpec);
     final CompactionTask taskCreatedWithTransformSpec = builder.build();
@@ -449,7 +449,7 @@ public void testCreateCompactionTaskWithMetricsSpec()
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.tuningConfig(createTuningConfig());
     builder.metricsSpec(aggregatorFactories);
     final CompactionTask taskCreatedWithTransformSpec = builder.build();
@@ -466,7 +466,7 @@ public void testCreateCompactionTaskWithNullSegmentGranularityInGranularitySpecA
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.tuningConfig(createTuningConfig());
     builder.segmentGranularity(Granularities.HOUR);
     builder.granularitySpec(new ClientCompactionTaskGranularitySpec(null, Granularities.DAY, null));
@@ -494,7 +494,7 @@ public void testCreateCompactionTaskWithSameGranularitySpecAndSegmentGranularity
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.tuningConfig(createTuningConfig());
     builder.segmentGranularity(Granularities.HOUR);
     builder.granularitySpec(new ClientCompactionTaskGranularitySpec(Granularities.HOUR, Granularities.DAY, null));
@@ -511,7 +511,7 @@ public void testSerdeWithInterval() throws IOException
     );
     final CompactionTask task = builder
         .inputSpec(
-            new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS))
+            new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null)
         )
         .tuningConfig(createTuningConfig())
         .context(ImmutableMap.of("testKey", "testContext"))
@@ -666,7 +666,7 @@ public void testInputSourceResources()
     );
     final CompactionTask task = builder
         .inputSpec(
-            new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS))
+            new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null)
         )
         .tuningConfig(createTuningConfig())
         .context(ImmutableMap.of("testKey", "testContext"))
@@ -770,7 +770,7 @@ public void testSegmentProviderFindSegmentsWithEmptySegmentsThrowException()
   {
     final SegmentProvider provider = new SegmentProvider(
         "datasource",
-        new CompactionIntervalSpec(Intervals.of("2021-01-01/P1D"), null)
+        new CompactionIntervalSpec(Intervals.of("2021-01-01/P1D"), null, null)
     );
 
     expectedException.expect(IllegalStateException.class);
@@ -784,9 +784,10 @@ public void testSegmentProviderFindSegmentsWithEmptySegmentsThrowException()
   public void testCreateIngestionSchema() throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -847,9 +848,10 @@ public void testCreateIngestionSchemaWithTargetPartitionSize() throws IOExceptio
         .build();
 
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -911,9 +913,10 @@ public void testCreateIngestionSchemaWithMaxTotalRows() throws IOException
         .build();
 
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -976,9 +979,10 @@ public void testCreateIngestionSchemaWithNumShards() throws IOException
         .build();
 
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1048,9 +1052,10 @@ public void testCreateIngestionSchemaWithCustomDimensionsSpec() throws IOExcepti
     );
 
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         customSpec,
         null,
         null,
@@ -1100,9 +1105,10 @@ public void testCreateIngestionSchemaWithCustomMetricsSpec() throws IOException
     };
 
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         customMetricsSpec,
@@ -1145,9 +1151,10 @@ public void testCreateIngestionSchemaWithCustomMetricsSpec() throws IOException
   public void testCreateIngestionSchemaWithCustomSegments() throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1197,6 +1204,7 @@ public void testCreateIngestionSchemaWithDifferentSegmentSet() throws IOExceptio
     // Remove one segment in the middle
     segments.remove(segments.size() / 2);
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
         new SegmentProvider(DATA_SOURCE, SpecificSegmentsSpec.fromSegments(segments)),
@@ -1229,6 +1237,7 @@ public void testMissingMetadata() throws IOException
     indexIO.removeMetadata(Iterables.getFirst(indexIO.getQueryableIndexMap().keySet(), null));
     final List<DataSegment> segments = new ArrayList<>(SEGMENTS);
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
         new SegmentProvider(DATA_SOURCE, SpecificSegmentsSpec.fromSegments(segments)),
@@ -1272,9 +1281,10 @@ public void testEmptyInterval()
   public void testSegmentGranularityAndNullQueryGranularity() throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1318,9 +1328,10 @@ public void testSegmentGranularityAndNullQueryGranularity() throws IOException
   public void testQueryGranularityAndNullSegmentGranularity() throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1361,9 +1372,10 @@ public void testQueryGranularityAndNullSegmentGranularity() throws IOException
   public void testQueryGranularityAndSegmentGranularityNonNull() throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1414,9 +1426,10 @@ public void testQueryGranularityAndSegmentGranularityNonNull() throws IOExceptio
   public void testNullGranularitySpec() throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1461,9 +1474,10 @@ public void testGranularitySpecWithNullQueryGranularityAndNullSegmentGranularity
       throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1508,9 +1522,10 @@ public void testGranularitySpecWithNotNullRollup()
       throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1540,9 +1555,10 @@ public void testGranularitySpecWithNullRollup()
       throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1574,9 +1590,10 @@ public void testMultiValuedDimensionsProcessing()
       throws IOException
   {
     final Map<Interval, DataSchema> dataSchemasForIntervals = CompactionTask.createDataSchemasForIntervals(
+        DATA_SOURCE,
         toolbox,
         LockGranularity.TIME_CHUNK,
-        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null)),
+        new SegmentProvider(DATA_SOURCE, new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null)),
         null,
         null,
         null,
@@ -1598,7 +1615,7 @@ public void testMSQRollupWithNoDimensionsSpecNeedsMVDInfo()
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.compactionRunner(new TestMSQCompactionRunner());
     final CompactionTask compactionTask = builder.build();
     // granularitySpec=null should assume a possible rollup
@@ -1612,7 +1629,7 @@ public void testMSQRangePartitionWithNoDimensionsSpecNeedsMVDInfo()
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.compactionRunner(new TestMSQCompactionRunner());
     builder.tuningConfig(TuningConfigBuilder.forCompactionTask()
                                             .withForceGuaranteedRollup(true)
@@ -1636,7 +1653,7 @@ public void testMSQRollupOnStringNeedsMVDInfo()
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.compactionRunner(new TestMSQCompactionRunner());
     builder.granularitySpec(new ClientCompactionTaskGranularitySpec(null, null, true));
 
@@ -1654,7 +1671,7 @@ public void testMSQRangePartitionOnStringNeedsMVDInfo()
         DATA_SOURCE,
         segmentCacheManagerFactory
     );
-    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS)));
+    builder.inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, SegmentUtils.hashIds(SEGMENTS), null));
     builder.compactionRunner(new TestMSQCompactionRunner());
 
     DimensionSchema stringDim = new StringDimensionSchema("string_dim_1", null, null);
diff --git a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
index ef417a8b838f..407e02ddb4e0 100644
--- a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
+++ b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/MSQCompactionRunnerTest.java
@@ -688,7 +688,7 @@ private CompactionTask createCompactionTask(
     context.putAll(contextParams);
 
     builder
-        .inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, null))
+        .inputSpec(new CompactionIntervalSpec(COMPACTION_INTERVAL, null, null))
         .tuningConfig(createTuningConfig(
             indexSpec,
             partitionsSpec == null ? new DynamicPartitionsSpec(100, null) : partitionsSpec