diff --git a/core/trino-main/src/main/java/io/trino/operator/output/BytePositionsAppender.java b/core/trino-main/src/main/java/io/trino/operator/output/BytePositionsAppender.java
new file mode 100644
index 000000000000..b272e0159748
--- /dev/null
+++ b/core/trino-main/src/main/java/io/trino/operator/output/BytePositionsAppender.java
@@ -0,0 +1,44 @@
+/*
+ * 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.output;
+
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+
+public class BytePositionsAppender
+        implements PositionsAppender
+{
+    @Override
+    public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
+    {
+        int[] positionArray = positions.elements();
+        if (block.mayHaveNull()) {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                if (block.isNull(position)) {
+                    blockBuilder.appendNull();
+                }
+                else {
+                    blockBuilder.writeByte(block.getByte(position, 0)).closeEntry();
+                }
+            }
+        }
+        else {
+            for (int i = 0; i < positions.size(); i++) {
+                blockBuilder.writeByte(block.getByte(positionArray[i], 0)).closeEntry();
+            }
+        }
+    }
+}
diff --git a/core/trino-main/src/main/java/io/trino/operator/output/Int128PositionsAppender.java b/core/trino-main/src/main/java/io/trino/operator/output/Int128PositionsAppender.java
new file mode 100644
index 000000000000..6e2e1ea12ecc
--- /dev/null
+++ b/core/trino-main/src/main/java/io/trino/operator/output/Int128PositionsAppender.java
@@ -0,0 +1,51 @@
+/*
+ * 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.output;
+
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+
+import static io.airlift.slice.SizeOf.SIZE_OF_LONG;
+
+public class Int128PositionsAppender
+        implements PositionsAppender
+{
+    @Override
+    public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
+    {
+        int[] positionArray = positions.elements();
+        if (block.mayHaveNull()) {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                if (block.isNull(position)) {
+                    blockBuilder.appendNull();
+                }
+                else {
+                    blockBuilder.writeLong(block.getLong(position, 0));
+                    blockBuilder.writeLong(block.getLong(position, SIZE_OF_LONG));
+                    blockBuilder.closeEntry();
+                }
+            }
+        }
+        else {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                blockBuilder.writeLong(block.getLong(position, 0));
+                blockBuilder.writeLong(block.getLong(position, SIZE_OF_LONG));
+                blockBuilder.closeEntry();
+            }
+        }
+    }
+}
diff --git a/core/trino-main/src/main/java/io/trino/operator/output/Int96PositionsAppender.java b/core/trino-main/src/main/java/io/trino/operator/output/Int96PositionsAppender.java
new file mode 100644
index 000000000000..572bbaf3a73c
--- /dev/null
+++ b/core/trino-main/src/main/java/io/trino/operator/output/Int96PositionsAppender.java
@@ -0,0 +1,51 @@
+/*
+ * 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.output;
+
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+
+import static io.airlift.slice.SizeOf.SIZE_OF_LONG;
+
+public class Int96PositionsAppender
+        implements PositionsAppender
+{
+    @Override
+    public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
+    {
+        int[] positionArray = positions.elements();
+        if (block.mayHaveNull()) {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                if (block.isNull(position)) {
+                    blockBuilder.appendNull();
+                }
+                else {
+                    blockBuilder.writeLong(block.getLong(position, 0));
+                    blockBuilder.writeInt(block.getInt(position, SIZE_OF_LONG));
+                    blockBuilder.closeEntry();
+                }
+            }
+        }
+        else {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                blockBuilder.writeLong(block.getLong(position, 0));
+                blockBuilder.writeInt(block.getInt(position, SIZE_OF_LONG));
+                blockBuilder.closeEntry();
+            }
+        }
+    }
+}
diff --git a/core/trino-main/src/main/java/io/trino/operator/output/IntPositionsAppender.java b/core/trino-main/src/main/java/io/trino/operator/output/IntPositionsAppender.java
new file mode 100644
index 000000000000..70a9b3dc3b40
--- /dev/null
+++ b/core/trino-main/src/main/java/io/trino/operator/output/IntPositionsAppender.java
@@ -0,0 +1,44 @@
+/*
+ * 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.output;
+
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+
+public class IntPositionsAppender
+        implements PositionsAppender
+{
+    @Override
+    public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
+    {
+        int[] positionArray = positions.elements();
+        if (block.mayHaveNull()) {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                if (block.isNull(position)) {
+                    blockBuilder.appendNull();
+                }
+                else {
+                    blockBuilder.writeInt(block.getInt(position, 0)).closeEntry();
+                }
+            }
+        }
+        else {
+            for (int i = 0; i < positions.size(); i++) {
+                blockBuilder.writeInt(block.getInt(positionArray[i], 0)).closeEntry();
+            }
+        }
+    }
+}
diff --git a/core/trino-main/src/main/java/io/trino/operator/output/LongPositionsAppender.java b/core/trino-main/src/main/java/io/trino/operator/output/LongPositionsAppender.java
new file mode 100644
index 000000000000..8cd41378838a
--- /dev/null
+++ b/core/trino-main/src/main/java/io/trino/operator/output/LongPositionsAppender.java
@@ -0,0 +1,44 @@
+/*
+ * 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.output;
+
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+
+public class LongPositionsAppender
+        implements PositionsAppender
+{
+    @Override
+    public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
+    {
+        int[] positionArray = positions.elements();
+        if (block.mayHaveNull()) {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                if (block.isNull(position)) {
+                    blockBuilder.appendNull();
+                }
+                else {
+                    blockBuilder.writeLong(block.getLong(position, 0)).closeEntry();
+                }
+            }
+        }
+        else {
+            for (int i = 0; i < positions.size(); i++) {
+                blockBuilder.writeLong(block.getLong(positionArray[i], 0)).closeEntry();
+            }
+        }
+    }
+}
diff --git a/core/trino-main/src/main/java/io/trino/operator/output/PositionsAppender.java b/core/trino-main/src/main/java/io/trino/operator/output/PositionsAppender.java
index 0b5aa8414f8b..ccdd77489968 100644
--- a/core/trino-main/src/main/java/io/trino/operator/output/PositionsAppender.java
+++ b/core/trino-main/src/main/java/io/trino/operator/output/PositionsAppender.java
@@ -24,24 +24,4 @@
 public interface PositionsAppender
 {
     void appendTo(IntArrayList positions, Block source, BlockBuilder target);
-
-    class TypedPositionsAppender
-            implements PositionsAppender
-    {
-        private final Type type;
-
-        public TypedPositionsAppender(Type type)
-        {
-            this.type = requireNonNull(type, "type is null");
-        }
-
-        @Override
-        public void appendTo(IntArrayList positions, Block source, BlockBuilder target)
-        {
-            int[] positionArray = positions.elements();
-            for (int i = 0; i < positions.size(); i++) {
-                type.appendTo(source, positionArray[i], target);
-            }
-        }
-    }
 }
diff --git a/core/trino-main/src/main/java/io/trino/operator/output/PositionsAppenderFactory.java b/core/trino-main/src/main/java/io/trino/operator/output/PositionsAppenderFactory.java
index 41021d98b70e..42a461b6b587 100644
--- a/core/trino-main/src/main/java/io/trino/operator/output/PositionsAppenderFactory.java
+++ b/core/trino-main/src/main/java/io/trino/operator/output/PositionsAppenderFactory.java
@@ -17,21 +17,17 @@
 import com.google.common.cache.CacheLoader;
 import io.airlift.bytecode.DynamicClassLoader;
 import io.trino.collect.cache.NonEvictableLoadingCache;
-import io.trino.operator.output.PositionsAppender.TypedPositionsAppender;
 import io.trino.spi.block.Block;
-import io.trino.spi.block.BlockBuilder;
 import io.trino.spi.block.Int128ArrayBlock;
 import io.trino.spi.block.Int96ArrayBlock;
 import io.trino.spi.type.FixedWidthType;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.VariableWidthType;
 import io.trino.sql.gen.IsolatedClass;
-import it.unimi.dsi.fastutil.ints.IntArrayList;
 
 import java.util.Objects;
 import java.util.Optional;
 
-import static io.airlift.slice.SizeOf.SIZE_OF_LONG;
 import static io.trino.collect.cache.SafeCaches.buildNonEvictableCache;
 import static java.util.Objects.requireNonNull;
 
@@ -125,201 +121,6 @@ private Class<? extends PositionsAppender> isolateAppenderClass(Class<? extends
         return isolatedBatchPositionsTransferClass;
     }
 
-    public static class LongPositionsAppender
-            implements PositionsAppender
-    {
-        @Override
-        public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
-        {
-            int[] positionArray = positions.elements();
-            if (block.mayHaveNull()) {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    if (block.isNull(position)) {
-                        blockBuilder.appendNull();
-                    }
-                    else {
-                        blockBuilder.writeLong(block.getLong(position, 0)).closeEntry();
-                    }
-                }
-            }
-            else {
-                for (int i = 0; i < positions.size(); i++) {
-                    blockBuilder.writeLong(block.getLong(positionArray[i], 0)).closeEntry();
-                }
-            }
-        }
-    }
-
-    public static class IntPositionsAppender
-            implements PositionsAppender
-    {
-        @Override
-        public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
-        {
-            int[] positionArray = positions.elements();
-            if (block.mayHaveNull()) {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    if (block.isNull(position)) {
-                        blockBuilder.appendNull();
-                    }
-                    else {
-                        blockBuilder.writeInt(block.getInt(position, 0)).closeEntry();
-                    }
-                }
-            }
-            else {
-                for (int i = 0; i < positions.size(); i++) {
-                    blockBuilder.writeInt(block.getInt(positionArray[i], 0)).closeEntry();
-                }
-            }
-        }
-    }
-
-    public static class BytePositionsAppender
-            implements PositionsAppender
-    {
-        @Override
-        public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
-        {
-            int[] positionArray = positions.elements();
-            if (block.mayHaveNull()) {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    if (block.isNull(position)) {
-                        blockBuilder.appendNull();
-                    }
-                    else {
-                        blockBuilder.writeByte(block.getByte(position, 0)).closeEntry();
-                    }
-                }
-            }
-            else {
-                for (int i = 0; i < positions.size(); i++) {
-                    blockBuilder.writeByte(block.getByte(positionArray[i], 0)).closeEntry();
-                }
-            }
-        }
-    }
-
-    public static class SlicePositionsAppender
-            implements PositionsAppender
-    {
-        @Override
-        public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
-        {
-            int[] positionArray = positions.elements();
-            if (block.mayHaveNull()) {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    if (block.isNull(position)) {
-                        blockBuilder.appendNull();
-                    }
-                    else {
-                        block.writeBytesTo(position, 0, block.getSliceLength(position), blockBuilder);
-                        blockBuilder.closeEntry();
-                    }
-                }
-            }
-            else {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    block.writeBytesTo(position, 0, block.getSliceLength(position), blockBuilder);
-                    blockBuilder.closeEntry();
-                }
-            }
-        }
-    }
-
-    public static class SmallintPositionsAppender
-            implements PositionsAppender
-    {
-        @Override
-        public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
-        {
-            int[] positionArray = positions.elements();
-            if (block.mayHaveNull()) {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    if (block.isNull(position)) {
-                        blockBuilder.appendNull();
-                    }
-                    else {
-                        blockBuilder.writeShort(block.getShort(position, 0)).closeEntry();
-                    }
-                }
-            }
-            else {
-                for (int i = 0; i < positions.size(); i++) {
-                    blockBuilder.writeShort(block.getShort(positionArray[i], 0)).closeEntry();
-                }
-            }
-        }
-    }
-
-    public static class Int96PositionsAppender
-            implements PositionsAppender
-    {
-        @Override
-        public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
-        {
-            int[] positionArray = positions.elements();
-            if (block.mayHaveNull()) {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    if (block.isNull(position)) {
-                        blockBuilder.appendNull();
-                    }
-                    else {
-                        blockBuilder.writeLong(block.getLong(position, 0));
-                        blockBuilder.writeInt(block.getInt(position, SIZE_OF_LONG));
-                        blockBuilder.closeEntry();
-                    }
-                }
-            }
-            else {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    blockBuilder.writeLong(block.getLong(position, 0));
-                    blockBuilder.writeInt(block.getInt(position, SIZE_OF_LONG));
-                    blockBuilder.closeEntry();
-                }
-            }
-        }
-    }
-
-    public static class Int128PositionsAppender
-            implements PositionsAppender
-    {
-        @Override
-        public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
-        {
-            int[] positionArray = positions.elements();
-            if (block.mayHaveNull()) {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    if (block.isNull(position)) {
-                        blockBuilder.appendNull();
-                    }
-                    else {
-                        blockBuilder.writeLong(block.getLong(position, 0));
-                        blockBuilder.writeLong(block.getLong(position, SIZE_OF_LONG));
-                        blockBuilder.closeEntry();
-                    }
-                }
-            }
-            else {
-                for (int i = 0; i < positions.size(); i++) {
-                    int position = positionArray[i];
-                    blockBuilder.writeLong(block.getLong(position, 0));
-                    blockBuilder.writeLong(block.getLong(position, SIZE_OF_LONG));
-                    blockBuilder.closeEntry();
-                }
-            }
-        }
-    }
-
     private static class CacheKey
     {
         private final Type type;
diff --git a/core/trino-main/src/main/java/io/trino/operator/output/SlicePositionsAppender.java b/core/trino-main/src/main/java/io/trino/operator/output/SlicePositionsAppender.java
new file mode 100644
index 000000000000..86bc96a76a0b
--- /dev/null
+++ b/core/trino-main/src/main/java/io/trino/operator/output/SlicePositionsAppender.java
@@ -0,0 +1,47 @@
+/*
+ * 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.output;
+
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+
+public class SlicePositionsAppender
+        implements PositionsAppender
+{
+    @Override
+    public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
+    {
+        int[] positionArray = positions.elements();
+        if (block.mayHaveNull()) {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                if (block.isNull(position)) {
+                    blockBuilder.appendNull();
+                }
+                else {
+                    block.writeBytesTo(position, 0, block.getSliceLength(position), blockBuilder);
+                    blockBuilder.closeEntry();
+                }
+            }
+        }
+        else {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                block.writeBytesTo(position, 0, block.getSliceLength(position), blockBuilder);
+                blockBuilder.closeEntry();
+            }
+        }
+    }
+}
diff --git a/core/trino-main/src/main/java/io/trino/operator/output/SmallintPositionsAppender.java b/core/trino-main/src/main/java/io/trino/operator/output/SmallintPositionsAppender.java
new file mode 100644
index 000000000000..c9708cd1e348
--- /dev/null
+++ b/core/trino-main/src/main/java/io/trino/operator/output/SmallintPositionsAppender.java
@@ -0,0 +1,44 @@
+/*
+ * 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.output;
+
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+
+public class SmallintPositionsAppender
+        implements PositionsAppender
+{
+    @Override
+    public void appendTo(IntArrayList positions, Block block, BlockBuilder blockBuilder)
+    {
+        int[] positionArray = positions.elements();
+        if (block.mayHaveNull()) {
+            for (int i = 0; i < positions.size(); i++) {
+                int position = positionArray[i];
+                if (block.isNull(position)) {
+                    blockBuilder.appendNull();
+                }
+                else {
+                    blockBuilder.writeShort(block.getShort(position, 0)).closeEntry();
+                }
+            }
+        }
+        else {
+            for (int i = 0; i < positions.size(); i++) {
+                blockBuilder.writeShort(block.getShort(positionArray[i], 0)).closeEntry();
+            }
+        }
+    }
+}
diff --git a/core/trino-main/src/main/java/io/trino/operator/output/TypedPositionsAppender.java b/core/trino-main/src/main/java/io/trino/operator/output/TypedPositionsAppender.java
new file mode 100644
index 000000000000..3e20eb0bb40b
--- /dev/null
+++ b/core/trino-main/src/main/java/io/trino/operator/output/TypedPositionsAppender.java
@@ -0,0 +1,41 @@
+/*
+ * 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.output;
+
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.type.Type;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+
+import static java.util.Objects.requireNonNull;
+
+public class TypedPositionsAppender
+        implements PositionsAppender
+{
+    private final Type type;
+
+    public TypedPositionsAppender(Type type)
+    {
+        this.type = requireNonNull(type, "type is null");
+    }
+
+    @Override
+    public void appendTo(IntArrayList positions, Block source, BlockBuilder target)
+    {
+        int[] positionArray = positions.elements();
+        for (int i = 0; i < positions.size(); i++) {
+            type.appendTo(source, positionArray[i], target);
+        }
+    }
+}