Reduce Lifespan object to a taskWide singleton

After the removal of grouped execution it is impossible for this object to be
anything else than the "taskWide" singleton. This object is at that point
completely useless and will exist just until its last occurrence is removed,
which will take place in the following commits.

Author: skrzypo987

core/trino-main/src/main/java/io/trino/execution/Lifespan.java

@@ -18,83 +18,49 @@
 import com.fasterxml.jackson.annotation.JsonValue;
 import org.openjdk.jol.info.ClassLayout;
 
-import java.util.Objects;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkState;
-import static java.lang.Integer.parseInt;
-
 public class Lifespan
 {
     private static final int INSTANCE_SIZE = ClassLayout.parseClass(Lifespan.class).instanceSize();
 
-    private static final Lifespan TASK_WIDE = new Lifespan(false, 0);
-
-    private final boolean grouped;
-    private final int groupId;
+    private static final Lifespan TASK_WIDE = new Lifespan();
 
     public static Lifespan taskWide()
     {
         return TASK_WIDE;
     }
 
-    public static Lifespan driverGroup(int id)
+    private Lifespan()
     {
-        return new Lifespan(true, id);
-    }
-
-    private Lifespan(boolean grouped, int groupId)
-    {
-        this.grouped = grouped;
-        this.groupId = groupId;
     }
 
     public boolean isTaskWide()
     {
-        return !grouped;
-    }
-
-    public int getId()
-    {
-        checkState(grouped);
-        return groupId;
+        return true;
     }
 
     @JsonCreator
     public static Lifespan jsonCreator(String value)
     {
-        if (value.equals("TaskWide")) {
-            return Lifespan.taskWide();
-        }
-        checkArgument(value.startsWith("Group"));
-        return Lifespan.driverGroup(parseInt(value.substring("Group".length())));
+        return Lifespan.taskWide();
     }
 
     @Override
     @JsonValue
     public String toString()
     {
-        return grouped ? "Group" + groupId : "TaskWide";
+        return "TaskWide";
     }
 
     @Override
     public boolean equals(Object o)
     {
-        if (this == o) {
-            return true;
-        }
-        if (o == null || getClass() != o.getClass()) {
-            return false;
-        }
-        Lifespan that = (Lifespan) o;
-        return grouped == that.grouped &&
-                groupId == that.groupId;
+        return true;
     }
 
     @Override
     public int hashCode()
     {
-        return Objects.hash(grouped, groupId);
+        return 42;
     }
 
     public long getRetainedSizeInBytes()

core/trino-main/src/main/java/io/trino/execution/scheduler/FixedSourcePartitionedScheduler.java

@@ -122,10 +122,7 @@ public FixedSourcePartitionedScheduler(
 
     private ConnectorPartitionHandle partitionHandleFor(Lifespan lifespan)
     {
-        if (lifespan.isTaskWide()) {
-            return NOT_PARTITIONED;
-        }
-        return partitionHandles.get(lifespan.getId());
+        return NOT_PARTITIONED;
     }
 
     @Override

core/trino-main/src/main/java/io/trino/execution/scheduler/SourcePartitionedScheduler.java

@@ -24,7 +24,6 @@
 import io.trino.execution.RemoteTask;
 import io.trino.execution.TableExecuteContext;
 import io.trino.execution.TableExecuteContextManager;
-import io.trino.execution.scheduler.FixedSourcePartitionedScheduler.BucketedSplitPlacementPolicy;
 import io.trino.metadata.InternalNode;
 import io.trino.metadata.Split;
 import io.trino.server.DynamicFilterService;
@@ -359,10 +358,6 @@ else if (pendingSplits.isEmpty()) {
             Multimap<InternalNode, Lifespan> noMoreSplitsNotification = ImmutableMultimap.of();
             if (pendingSplits.isEmpty() && scheduleGroup.state == ScheduleGroupState.NO_MORE_SPLITS) {
                 scheduleGroup.state = ScheduleGroupState.DONE;
-                if (!lifespan.isTaskWide()) {
-                    InternalNode node = ((BucketedSplitPlacementPolicy) splitPlacementPolicy).getNodeForBucket(lifespan.getId());
-                    noMoreSplitsNotification = ImmutableMultimap.of(node, lifespan);
-                }
             }
 
             // assign the splits with successful placements

core/trino-main/src/main/java/io/trino/sql/planner/NodePartitioningManager.java

@@ -217,14 +217,11 @@ private ToIntFunction<Split> getSplitToBucket(Session session, PartitioningHandl
         return split -> {
             int bucket;
             if (split.getConnectorSplit() instanceof EmptySplit) {
-                bucket = split.getLifespan().isTaskWide() ? 0 : split.getLifespan().getId();
+                bucket = 0;
             }
             else {
                 bucket = splitBucketFunction.applyAsInt(split.getConnectorSplit());
             }
-            if (!split.getLifespan().isTaskWide()) {
-                checkArgument(split.getLifespan().getId() == bucket);
-            }
             return bucket;
         };
     }

core/trino-main/src/test/java/io/trino/operator/TestDriverStats.java

@@ -30,7 +30,7 @@
 public class TestDriverStats
 {
     public static final DriverStats EXPECTED = new DriverStats(
-            Lifespan.driverGroup(21),
+            Lifespan.taskWide(),
 
             new DateTime(1),
             new DateTime(2),
@@ -86,7 +86,7 @@ public void testJson()
 
     public static void assertExpectedDriverStats(DriverStats actual)
     {
-        assertEquals(actual.getLifespan(), Lifespan.driverGroup(21));
+        assertEquals(actual.getLifespan(), Lifespan.taskWide());
 
         assertEquals(actual.getCreateTime(), new DateTime(1, UTC));
         assertEquals(actual.getStartTime(), new DateTime(2, UTC));

core/trino-main/src/test/java/io/trino/operator/exchange/TestLocalExchange.java

@@ -671,7 +671,7 @@ public void testMismatchedExecutionStrategy()
                 Optional.empty(),
                 LOCAL_EXCHANGE_MAX_BUFFERED_BYTES,
                 TYPE_OPERATOR_FACTORY);
-        assertThatThrownBy(() -> ungroupedLocalExchangeFactory.getLocalExchange(Lifespan.driverGroup(3)))
+        assertThatThrownBy(() -> ungroupedLocalExchangeFactory.getLocalExchange(Lifespan.taskWide()))
                 .isInstanceOf(IllegalArgumentException.class)
                 .hasMessage("LocalExchangeFactory is declared as UNGROUPED_EXECUTION. Driver-group exchange cannot be created.");
 

core/trino-main/src/test/java/io/trino/server/remotetask/TestHttpRemoteTask.java

@@ -183,12 +183,11 @@ public void testRegular()
         testingTaskResource.setInitialTaskInfo(remoteTask.getTaskInfo());
         remoteTask.start();
 
-        Lifespan lifespan = Lifespan.driverGroup(3);
-        remoteTask.addSplits(ImmutableMultimap.of(TABLE_SCAN_NODE_ID, new Split(new CatalogName("test"), TestingSplit.createLocalSplit(), lifespan)));
+        remoteTask.addSplits(ImmutableMultimap.of(TABLE_SCAN_NODE_ID, new Split(new CatalogName("test"), TestingSplit.createLocalSplit(), Lifespan.taskWide())));
         poll(() -> testingTaskResource.getTaskSplitAssignment(TABLE_SCAN_NODE_ID) != null);
         poll(() -> testingTaskResource.getTaskSplitAssignment(TABLE_SCAN_NODE_ID).getSplits().size() == 1);
 
-        remoteTask.noMoreSplits(TABLE_SCAN_NODE_ID, lifespan);
+        remoteTask.noMoreSplits(TABLE_SCAN_NODE_ID, Lifespan.taskWide());
         poll(() -> testingTaskResource.getTaskSplitAssignment(TABLE_SCAN_NODE_ID).getNoMoreSplitsForLifespan().size() == 1);
 
         remoteTask.noMoreSplits(TABLE_SCAN_NODE_ID);
@@ -413,8 +412,7 @@ private void runTest(FailureScenario failureScenario)
     private void addSplit(RemoteTask remoteTask, TestingTaskResource testingTaskResource, int expectedSplitsCount)
             throws InterruptedException
     {
-        Lifespan lifespan = Lifespan.driverGroup(3);
-        remoteTask.addSplits(ImmutableMultimap.of(TABLE_SCAN_NODE_ID, new Split(new CatalogName("test"), TestingSplit.createLocalSplit(), lifespan)));
+        remoteTask.addSplits(ImmutableMultimap.of(TABLE_SCAN_NODE_ID, new Split(new CatalogName("test"), TestingSplit.createLocalSplit(), Lifespan.taskWide())));
         // wait for splits to be received by remote task
         poll(() -> testingTaskResource.getTaskSplitAssignment(TABLE_SCAN_NODE_ID) != null);
         poll(() -> testingTaskResource.getTaskSplitAssignment(TABLE_SCAN_NODE_ID).getSplits().size() == expectedSplitsCount);