Move spilling join operator classes to subpackage

core/trino-main/src/main/java/io/trino/operator/JoinOperatorType.java

@@ -13,18 +13,17 @@
  */
 package io.trino.operator;
 
-import io.trino.operator.join.LookupJoinOperatorFactory;
 import io.trino.sql.planner.plan.JoinType;
 
-import static io.trino.operator.join.LookupJoinOperatorFactory.JoinType.FULL_OUTER;
-import static io.trino.operator.join.LookupJoinOperatorFactory.JoinType.INNER;
-import static io.trino.operator.join.LookupJoinOperatorFactory.JoinType.LOOKUP_OUTER;
-import static io.trino.operator.join.LookupJoinOperatorFactory.JoinType.PROBE_OUTER;
+import static io.trino.operator.join.JoinType.FULL_OUTER;
+import static io.trino.operator.join.JoinType.INNER;
+import static io.trino.operator.join.JoinType.LOOKUP_OUTER;
+import static io.trino.operator.join.JoinType.PROBE_OUTER;
 import static java.util.Objects.requireNonNull;
 
 public class JoinOperatorType
 {
-    private final LookupJoinOperatorFactory.JoinType type;
+    private final io.trino.operator.join.JoinType type;
     private final boolean outputSingleMatch;
     private final boolean waitForBuild;
 
@@ -58,7 +57,7 @@ public static JoinOperatorType fullOuterJoin()
         return new JoinOperatorType(FULL_OUTER, false, false);
     }
 
-    private JoinOperatorType(LookupJoinOperatorFactory.JoinType type, boolean outputSingleMatch, boolean waitForBuild)
+    private JoinOperatorType(io.trino.operator.join.JoinType type, boolean outputSingleMatch, boolean waitForBuild)
     {
         this.type = requireNonNull(type, "type is null");
         this.outputSingleMatch = outputSingleMatch;
@@ -75,7 +74,7 @@ public boolean isWaitForBuild()
         return waitForBuild;
     }
 
-    public LookupJoinOperatorFactory.JoinType getType()
+    public io.trino.operator.join.JoinType getType()
     {
         return type;
     }

core/trino-main/src/main/java/io/trino/operator/OperatorFactories.java

@@ -14,9 +14,9 @@
 package io.trino.operator;
 
 import io.trino.operator.join.JoinBridgeManager;
-import io.trino.operator.join.JoinProbe.JoinProbeFactory;
-import io.trino.operator.join.LookupJoinOperatorFactory;
 import io.trino.operator.join.LookupSourceFactory;
+import io.trino.operator.join.spilling.JoinProbe.JoinProbeFactory;
+import io.trino.operator.join.spilling.LookupJoinOperatorFactory;
 import io.trino.operator.join.unspilled.JoinProbe;
 import io.trino.operator.join.unspilled.PartitionedLookupSourceFactory;
 import io.trino.spi.type.Type;

core/trino-main/src/main/java/io/trino/operator/WorkProcessorOperatorAdapter.java

@@ -17,7 +17,7 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import io.trino.memory.context.MemoryTrackingContext;
 import io.trino.operator.join.JoinOperatorFactory;
-import io.trino.operator.join.LookupJoinOperatorFactory;
+import io.trino.operator.join.spilling.LookupJoinOperatorFactory;
 import io.trino.spi.Page;
 
 import java.util.Optional;

core/trino-main/src/main/java/io/trino/operator/join/DefaultPageJoiner.java

@@ -24,8 +24,9 @@
 import io.trino.operator.SpillMetrics;
 import io.trino.operator.WorkProcessor;
 import io.trino.operator.exchange.LocalPartitionGenerator;
-import io.trino.operator.join.JoinProbe.JoinProbeFactory;
-import io.trino.operator.join.LookupJoinOperatorFactory.JoinType;
+import io.trino.operator.join.spilling.JoinProbe;
+import io.trino.operator.join.spilling.JoinProbe.JoinProbeFactory;
+import io.trino.operator.join.spilling.LookupJoinPageBuilder;
 import io.trino.spi.Page;
 import io.trino.spi.type.Type;
 import io.trino.spiller.PartitioningSpiller;
@@ -55,9 +56,9 @@
 import static io.trino.operator.WorkProcessor.TransformationState.needsMoreData;
 import static io.trino.operator.WorkProcessor.TransformationState.ofResult;
 import static io.trino.operator.WorkProcessor.TransformationState.yielded;
-import static io.trino.operator.join.LookupJoinOperatorFactory.JoinType.FULL_OUTER;
-import static io.trino.operator.join.LookupJoinOperatorFactory.JoinType.PROBE_OUTER;
-import static io.trino.operator.join.PartitionedLookupSourceFactory.NO_SPILL_EPOCH;
+import static io.trino.operator.join.JoinType.FULL_OUTER;
+import static io.trino.operator.join.JoinType.PROBE_OUTER;
+import static io.trino.operator.join.spilling.PartitionedLookupSourceFactory.NO_SPILL_EPOCH;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 

core/trino-main/src/main/java/io/trino/operator/join/JoinBridgeManager.java

@@ -18,6 +18,7 @@
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.trino.operator.ReferenceCount;
+import io.trino.operator.join.spilling.PartitionedLookupSourceFactory;
 import io.trino.spi.type.Type;
 
 import java.util.List;

core/trino-main/src/main/java/io/trino/operator/join/JoinOperatorInfo.java

@@ -16,7 +16,6 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.trino.operator.OperatorInfo;
-import io.trino.operator.join.LookupJoinOperatorFactory.JoinType;
 import io.trino.spi.Mergeable;
 
 import java.util.OptionalLong;

core/trino-main/src/main/java/io/trino/operator/join/JoinStatisticsCounter.java

@@ -14,7 +14,6 @@
 package io.trino.operator.join;
 
 import io.trino.operator.OperatorInfo;
-import io.trino.operator.join.LookupJoinOperatorFactory.JoinType;
 
 import java.util.OptionalLong;
 import java.util.function.Supplier;

core/trino-main/src/main/java/io/trino/operator/join/JoinType.java

@@ -0,0 +1,22 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.operator.join;
+
+public enum JoinType
+{
+    INNER,
+    PROBE_OUTER, // the Probe is the outer side of the join
+    LOOKUP_OUTER, // The LookupSource is the outer side of the join
+    FULL_OUTER,
+}

core/trino-main/src/main/java/io/trino/operator/join/LookupSourceFactory.java

@@ -15,6 +15,7 @@
 
 import com.google.common.util.concurrent.ListenableFuture;
 import io.trino.operator.TaskContext;
+import io.trino.operator.join.spilling.PartitionedConsumption;
 import io.trino.spi.type.Type;
 
 import java.util.List;

core/trino-main/src/main/java/io/trino/operator/join/HashBuilderOperator.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/HashBuilderOperator.java

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
@@ -31,6 +31,8 @@
 import io.trino.operator.OperatorFactory;
 import io.trino.operator.PagesIndex;
 import io.trino.operator.SpillMetrics;
+import io.trino.operator.join.JoinBridgeManager;
+import io.trino.operator.join.LookupSourceSupplier;
 import io.trino.spi.Page;
 import io.trino.spi.metrics.Metric;
 import io.trino.spi.metrics.Metrics;

core/trino-main/src/main/java/io/trino/operator/join/JoinProbe.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/JoinProbe.java

@@ -11,9 +11,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
 import com.google.common.primitives.Ints;
+import io.trino.operator.join.LookupSource;
 import io.trino.spi.Page;
 
 import java.util.List;

core/trino-main/src/main/java/io/trino/operator/join/LookupJoinOperator.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/LookupJoinOperator.java

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
 import com.google.common.util.concurrent.ListenableFuture;
 import io.trino.operator.HashGenerator;
@@ -20,9 +20,13 @@
 import io.trino.operator.SpillMetrics;
 import io.trino.operator.WorkProcessor;
 import io.trino.operator.WorkProcessorOperator;
-import io.trino.operator.join.JoinProbe.JoinProbeFactory;
-import io.trino.operator.join.LookupJoinOperatorFactory.JoinType;
+import io.trino.operator.join.DefaultPageJoiner;
+import io.trino.operator.join.JoinStatisticsCounter;
+import io.trino.operator.join.JoinType;
+import io.trino.operator.join.LookupSourceFactory;
+import io.trino.operator.join.LookupSourceProvider;
 import io.trino.operator.join.PageJoiner.PageJoinerFactory;
+import io.trino.operator.join.spilling.JoinProbe.JoinProbeFactory;
 import io.trino.spi.Page;
 import io.trino.spi.metrics.Metrics;
 import io.trino.spi.type.Type;

core/trino-main/src/main/java/io/trino/operator/join/LookupJoinOperatorFactory.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/LookupJoinOperatorFactory.java

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.primitives.Ints;
@@ -23,8 +23,12 @@
 import io.trino.operator.WorkProcessor;
 import io.trino.operator.WorkProcessorOperator;
 import io.trino.operator.WorkProcessorOperatorFactory;
-import io.trino.operator.join.JoinProbe.JoinProbeFactory;
+import io.trino.operator.join.JoinBridgeManager;
+import io.trino.operator.join.JoinOperatorFactory;
+import io.trino.operator.join.JoinType;
 import io.trino.operator.join.LookupOuterOperator.LookupOuterOperatorFactory;
+import io.trino.operator.join.LookupSourceFactory;
+import io.trino.operator.join.spilling.JoinProbe.JoinProbeFactory;
 import io.trino.spi.Page;
 import io.trino.spi.type.Type;
 import io.trino.spiller.PartitioningSpillerFactory;
@@ -38,21 +42,13 @@
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static io.trino.operator.InterpretedHashGenerator.createChannelsHashGenerator;
-import static io.trino.operator.join.LookupJoinOperatorFactory.JoinType.INNER;
-import static io.trino.operator.join.LookupJoinOperatorFactory.JoinType.PROBE_OUTER;
+import static io.trino.operator.join.JoinType.INNER;
+import static io.trino.operator.join.JoinType.PROBE_OUTER;
 import static java.util.Objects.requireNonNull;
 
 public class LookupJoinOperatorFactory
         implements JoinOperatorFactory, WorkProcessorOperatorFactory
 {
-    public enum JoinType
-    {
-        INNER,
-        PROBE_OUTER, // the Probe is the outer side of the join
-        LOOKUP_OUTER, // The LookupSource is the outer side of the join
-        FULL_OUTER,
-    }
-
     private final int operatorId;
     private final PlanNodeId planNodeId;
     private final List<Type> probeTypes;

core/trino-main/src/main/java/io/trino/operator/join/LookupJoinPageBuilder.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/LookupJoinPageBuilder.java

@@ -11,8 +11,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
+import io.trino.operator.join.LookupSource;
 import io.trino.spi.Page;
 import io.trino.spi.PageBuilder;
 import io.trino.spi.block.Block;

core/trino-main/src/main/java/io/trino/operator/join/PartitionedConsumption.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/PartitionedConsumption.java

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
 import com.google.common.collect.AbstractIterator;
 import com.google.common.collect.ImmutableList;
@@ -45,7 +45,7 @@ public final class PartitionedConsumption<T>
     @Nullable
     private List<Partition<T>> partitions;
 
-    PartitionedConsumption(
+    public PartitionedConsumption(
             int consumersCount,
             Iterable<Integer> partitionNumbers,
             IntFunction<ListenableFuture<T>> loader,
@@ -89,7 +89,7 @@ private List<Partition<T>> createPartitions(
         return partitions.build();
     }
 
-    Iterator<Partition<T>> beginConsumption()
+    public Iterator<Partition<T>> beginConsumption()
     {
         Queue<Partition<T>> partitions = new ArrayDeque<>(requireNonNull(this.partitions, "partitions is already null"));
         if (consumed.incrementAndGet() >= consumersCount) {

core/trino-main/src/main/java/io/trino/operator/join/PartitionedLookupSource.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/PartitionedLookupSource.java

@@ -11,13 +11,16 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
 import com.google.common.io.Closer;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
 import io.trino.annotation.NotThreadSafe;
 import io.trino.operator.NullSafeHashCompiler;
 import io.trino.operator.exchange.LocalPartitionGenerator;
+import io.trino.operator.join.LookupSource;
+import io.trino.operator.join.OuterPositionIterator;
+import io.trino.operator.join.TrackingLookupSourceSupplier;
 import io.trino.spi.Page;
 import io.trino.spi.PageBuilder;
 import io.trino.spi.type.Type;

core/trino-main/src/main/java/io/trino/operator/join/PartitionedLookupSourceFactory.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/PartitionedLookupSourceFactory.java

@@ -11,7 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -21,7 +21,12 @@
 import com.google.errorprone.annotations.concurrent.GuardedBy;
 import io.trino.annotation.NotThreadSafe;
 import io.trino.operator.NullSafeHashCompiler;
+import io.trino.operator.join.LookupSource;
+import io.trino.operator.join.LookupSourceFactory;
+import io.trino.operator.join.LookupSourceProvider;
 import io.trino.operator.join.LookupSourceProvider.LookupSourceLease;
+import io.trino.operator.join.OuterPositionIterator;
+import io.trino.operator.join.TrackingLookupSourceSupplier;
 import io.trino.spi.Page;
 import io.trino.spi.PageBuilder;
 import io.trino.spi.type.Type;
@@ -48,7 +53,7 @@
 import static com.google.common.util.concurrent.Futures.transform;
 import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
 import static io.trino.operator.join.OuterLookupSource.createOuterLookupSourceSupplier;
-import static io.trino.operator.join.PartitionedLookupSource.createPartitionedLookupSourceSupplier;
+import static io.trino.operator.join.spilling.PartitionedLookupSource.createPartitionedLookupSourceSupplier;
 import static java.util.Collections.emptyList;
 import static java.util.Objects.requireNonNull;
 

core/trino-main/src/main/java/io/trino/operator/join/SpilledLookupSourceHandle.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/SpilledLookupSourceHandle.java

@@ -11,13 +11,14 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.errorprone.annotations.ThreadSafe;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
+import io.trino.operator.join.LookupSource;
 import jakarta.annotation.Nullable;
 
 import java.util.function.Supplier;

core/trino-main/src/main/java/io/trino/operator/join/SpillingJoinProcessor.java → core/trino-main/src/main/java/io/trino/operator/join/spilling/SpillingJoinProcessor.java

@@ -11,15 +11,20 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package io.trino.operator.join;
+package io.trino.operator.join.spilling;
 
 import com.google.common.io.Closer;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.trino.operator.SpillMetrics;
 import io.trino.operator.WorkProcessor;
 import io.trino.operator.join.DefaultPageJoiner.SavedRow;
+import io.trino.operator.join.LookupSource;
+import io.trino.operator.join.LookupSourceFactory;
+import io.trino.operator.join.LookupSourceProvider;
+import io.trino.operator.join.PageJoiner;
 import io.trino.operator.join.PageJoiner.PageJoinerFactory;
+import io.trino.operator.join.StaticLookupSourceProvider;
 import io.trino.spi.Page;
 import io.trino.spiller.PartitioningSpillerFactory;
 import jakarta.annotation.Nullable;

core/trino-main/src/main/java/io/trino/operator/join/unspilled/HashBuilderOperator.java

@@ -42,7 +42,7 @@
 import static java.util.Objects.requireNonNull;
 
 /**
- * Like {@link io.trino.operator.join.HashBuilderOperator} but simplified,
+ * Like {@link io.trino.operator.join.spilling.HashBuilderOperator} but simplified,
  * without spill support.
  */
 @ThreadSafe

core/trino-main/src/main/java/io/trino/operator/join/unspilled/LookupJoinOperator.java

@@ -20,7 +20,7 @@
 import io.trino.operator.WorkProcessor;
 import io.trino.operator.WorkProcessorOperator;
 import io.trino.operator.join.JoinStatisticsCounter;
-import io.trino.operator.join.LookupJoinOperatorFactory.JoinType;
+import io.trino.operator.join.JoinType;
 import io.trino.operator.join.LookupSource;
 import io.trino.operator.join.unspilled.JoinProbe.JoinProbeFactory;
 import io.trino.spi.Page;
@@ -35,7 +35,7 @@
 import static java.util.Objects.requireNonNull;
 
 /**
- * Like {@link io.trino.operator.join.LookupJoinOperator} but simplified,
+ * Like {@link io.trino.operator.join.spilling.LookupJoinOperator} but simplified,
  * without spill support.
  */
 public class LookupJoinOperator