Nightly upstream merge

client/trino-cli/pom.xml

@@ -18,7 +18,7 @@
         <airstyle.rewriteUnusedLambdaParameters>false</airstyle.rewriteUnusedLambdaParameters>
         <checkstyle.violation.ignore>UnusedLambdaParameterShouldBeUnnamed,UseEnhancedSwitch</checkstyle.violation.ignore>
         <main-class>io.trino.cli.Trino</main-class>
-        <dep.jline.version>4.4.1</dep.jline.version>
+        <dep.jline.version>4.4.2</dep.jline.version>
     </properties>
 
     <dependencies>

client/trino-cli/src/main/java/io/trino/cli/StatusPrinter.java

@@ -22,15 +22,10 @@
 import io.trino.client.StatementStats;
 import org.jline.terminal.Attributes;
 import org.jline.terminal.Terminal;
-import org.jline.terminal.impl.AbstractUnixSysTerminal;
 import org.jline.utils.AttributedString;
 import org.jline.utils.AttributedStyle;
-import org.jline.utils.NonBlockingReader;
 
-import java.io.FileDescriptor;
-import java.io.FileInputStream;
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.PrintStream;
 import java.util.List;
 import java.util.OptionalInt;
@@ -64,16 +59,6 @@ public class StatusPrinter
     private static final int CTRL_C = 3;
     private static final int CTRL_P = 16;
 
-    // A timed read on terminal.reader() that expires leaves JLine's pump thread blocked
-    // in a read on the tty, where it steals the next keystroke typed into an external
-    // process spawned afterwards, such as the pager. None of the JLine stream wrappers
-    // report available(), so poll the stdin file descriptor directly and only issue a
-    // read once a byte is queued, so that it completes without parking the pump thread.
-    // This only applies to terminals reading the process stdin: AbstractUnixSysTerminal
-    // is created only when stdin is a tty and always reads FileDescriptor.in. Other
-    // terminals (Windows console, /dev/tty fallback, dumb) keep the timed read.
-    private static final InputStream RAW_TERMINAL_INPUT = new FileInputStream(FileDescriptor.in);
-
     private final long start = System.nanoTime();
     private final StatementClient client;
     private final PrintStream out;
@@ -486,14 +471,6 @@ private void reprintLine(String line)
     private static int readKey(Terminal terminal)
     {
         try {
-            if (terminal instanceof AbstractUnixSysTerminal) {
-                if (RAW_TERMINAL_INPUT.available() == 0) {
-                    return NonBlockingReader.READ_EXPIRED;
-                }
-                // a byte is queued, so the read returns immediately; the generous timeout
-                // only covers pump thread scheduling delays
-                return terminal.reader().read(100L);
-            }
             return terminal.reader().read(1L);
         }
         catch (IOException e) {

core/trino-main/src/main/java/io/trino/ExceededOutputLimitException.java

@@ -0,0 +1,28 @@
+/*
+ * 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;
+
+import io.airlift.units.DataSize;
+import io.trino.spi.TrinoException;
+
+import static io.trino.spi.StandardErrorCode.EXCEEDED_OUTPUT_LIMIT;
+
+public class ExceededOutputLimitException
+        extends TrinoException
+{
+    public ExceededOutputLimitException(DataSize limit)
+    {
+        super(EXCEEDED_OUTPUT_LIMIT, "Exceeded output data size limit of " + limit);
+    }
+}

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

@@ -24,6 +24,7 @@
 import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.context.Context;
 import io.trino.ExceededCpuLimitException;
+import io.trino.ExceededOutputLimitException;
 import io.trino.ExceededScanLimitException;
 import io.trino.ExceededWriteLimitException;
 import io.trino.Session;
@@ -80,6 +81,7 @@ public class QueryManager
     private final Duration maxQueryCpuTime;
     private final Optional<DataSize> maxQueryScanPhysicalBytes;
     private final Optional<DataSize> maxQueryWritePhysicalSize;
+    private final Optional<DataSize> maxQueryOutputDataSize;
 
     private final ExecutorService queryExecutor;
     private final ThreadPoolExecutorMBean queryExecutorMBean;
@@ -96,6 +98,7 @@ public QueryManager(ClusterMemoryManager memoryManager, Tracer tracer, QueryMana
         this.maxQueryCpuTime = queryManagerConfig.getQueryMaxCpuTime();
         this.maxQueryScanPhysicalBytes = queryManagerConfig.getQueryMaxScanPhysicalBytes();
         this.maxQueryWritePhysicalSize = queryManagerConfig.getQueryMaxWritePhysicalSize();
+        this.maxQueryOutputDataSize = queryManagerConfig.getQueryMaxOutputDataSize();
 
         this.queryExecutor = newCachedThreadPool(threadsNamed("query-scheduler-%s"));
         this.queryExecutorMBean = new ThreadPoolExecutorMBean((ThreadPoolExecutor) queryExecutor);
@@ -138,6 +141,13 @@ public void start()
             catch (Throwable e) {
                 log.error(e, "Error enforcing query write bytes limits");
             }
+
+            try {
+                enforceOutputDataSizeLimits();
+            }
+            catch (Throwable e) {
+                log.error(e, "Error enforcing query output data size limits");
+            }
         }, 1, 1, TimeUnit.SECONDS);
     }
 
@@ -456,4 +466,15 @@ private void enforceWriteLimits()
             });
         }
     }
+
+    private void enforceOutputDataSizeLimits()
+    {
+        maxQueryOutputDataSize.ifPresent(outputLimit -> {
+            for (QueryExecution query : queryTracker.getAllQueries()) {
+                if (query.getQueryInfo().getQueryStats().getOutputDataSize().compareTo(outputLimit) > 0) {
+                    query.fail(new ExceededOutputLimitException(outputLimit));
+                }
+            }
+        });
+    }
 }

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

@@ -103,6 +103,7 @@ public class QueryManagerConfig
     private Duration queryMaxCpuTime = new Duration(1_000_000_000, TimeUnit.DAYS);
     private Optional<DataSize> queryMaxScanPhysicalBytes = Optional.empty();
     private Optional<DataSize> queryMaxWritePhysicalSize = Optional.empty();
+    private Optional<DataSize> queryMaxOutputDataSize = Optional.empty();
     private int queryReportedRuleStatsLimit = 10;
     private int dispatcherQueryPoolSize = DISPATCHER_THREADPOOL_MAX_SIZE;
 
@@ -542,6 +543,19 @@ public QueryManagerConfig setQueryMaxWritePhysicalSize(DataSize queryMaxWritePhy
         return this;
     }
 
+    @NotNull
+    public Optional<@MinDataSize("1MB") DataSize> getQueryMaxOutputDataSize()
+    {
+        return queryMaxOutputDataSize;
+    }
+
+    @Config("query.max-output-data-size")
+    public QueryManagerConfig setQueryMaxOutputDataSize(DataSize queryMaxOutputDataSize)
+    {
+        this.queryMaxOutputDataSize = Optional.ofNullable(queryMaxOutputDataSize);
+        return this;
+    }
+
     @Min(1)
     public int getQueryReportedRuleStatsLimit()
     {

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

@@ -233,12 +233,17 @@ public void close()
     private final class PageAccounting
     {
         private static final int COMPACTION_MIN_FILL_MULTIPLIER = 2;
+        // Copy a page once when its backing arrays retain over 12.5% and at least 4 KB more than the data they hold;
+        // block builders size their arrays from the previous page, so retained pages commonly carry unused capacity
+        private static final int COMPACTION_MAX_SLACK_DIVISOR = 8;
+        private static final long COMPACTION_MIN_SLACK_BYTES = 4 * 1024;
 
         private final int pageId;
         private Page page;
         private long[] rowIds;
         // Start off locked to give the caller time to declare which rows to reference
         private boolean lockedPage = true;
+        private boolean compacted;
         private int activePositions;
 
         public PageAccounting(int pageId, Page page)
@@ -310,14 +315,31 @@ public boolean isPruneEligible()
         public boolean isCompactionEligible()
         {
             // Compaction is only allowed if the page is unlocked
-            return !lockedPage && activePositions * COMPACTION_MIN_FILL_MULTIPLIER < page.getPositionCount();
+            if (lockedPage) {
+                return false;
+            }
+            return activePositions * COMPACTION_MIN_FILL_MULTIPLIER < page.getPositionCount() || hasExcessRetainedBytes();
+        }
+
+        private boolean hasExcessRetainedBytes()
+        {
+            if (compacted) {
+                return false;
+            }
+            long sizeInBytes = page.getSizeInBytes();
+            long slackBytes = page.getRetainedSizeInBytes() - sizeInBytes;
+            return slackBytes > Math.max(sizeInBytes / COMPACTION_MAX_SLACK_DIVISOR, COMPACTION_MIN_SLACK_BYTES);
         }
 
         public void compact()
         {
             checkState(!lockedPage, "Should not attempt compaction when page is locked");
 
             if (activePositions == page.getPositionCount()) {
+                if (hasExcessRetainedBytes()) {
+                    page.compact();
+                    compacted = true;
+                }
                 return;
             }
 
@@ -338,11 +360,12 @@ public void compact()
             // Compact page
             page = page.copyPositions(positionsToKeep, 0, positionsToKeep.length);
             rowIds = newRowIds;
+            compacted = true;
         }
 
         public long sizeOf()
         {
-            return PAGE_ACCOUNTING_INSTANCE_SIZE + page.getSizeInBytes() + SizeOf.sizeOf(rowIds);
+            return PAGE_ACCOUNTING_INSTANCE_SIZE + page.getRetainedSizeInBytes() + SizeOf.sizeOf(rowIds);
         }
     }
 

core/trino-main/src/main/java/io/trino/operator/aggregation/DecimalAverageAggregation.java

@@ -160,7 +160,7 @@ public static Int128 average(LongDecimalWithOverflowAndLongState state, DecimalT
         long overflow = state.getOverflow();
         if (overflow != 0) {
             BigDecimal sum = new BigDecimal(Int128.valueOf(decimal[offset], decimal[offset + 1]).toBigInteger(), type.getScale());
-            sum = sum.add(new BigDecimal(OVERFLOW_MULTIPLIER.multiply(BigInteger.valueOf(overflow))));
+            sum = sum.add(new BigDecimal(OVERFLOW_MULTIPLIER.multiply(BigInteger.valueOf(overflow)), type.getScale()));
 
             BigDecimal count = BigDecimal.valueOf(state.getLong());
             return Decimals.encodeScaledValue(sum.divide(count, type.getScale(), HALF_UP), type.getScale());

core/trino-main/src/main/java/io/trino/operator/project/InputChannels.java

@@ -182,6 +182,11 @@ public Page getColumns(int[] channels)
         public void selectPositions(int[] positions, int offset, int size)
         {
             sourcePage.selectPositions(positions, offset, size);
+            selectLoadedBlocks(positions, offset, size);
+        }
+
+        private void selectLoadedBlocks(int[] positions, int offset, int size)
+        {
             for (int i = 0; i < blocks.length; i++) {
                 Block block = blocks[i];
                 if (block != null) {

core/trino-main/src/main/java/io/trino/operator/project/PageProcessor.java

@@ -135,9 +135,33 @@ public WorkProcessor<Page> createWorkProcessor(
             return WorkProcessor.of(new Page(selectedPositions.size()));
         }
 
+        if (!isAllPositions(selectedPositions, page.getPositionCount())) {
+            int[] positions;
+            int positionsOffset;
+            if (selectedPositions.isList()) {
+                positions = selectedPositions.getPositions();
+                positionsOffset = selectedPositions.getOffset();
+            }
+            else {
+                positions = new int[selectedPositions.size()];
+                positionsOffset = 0;
+                for (int index = 0; index < positions.length; index++) {
+                    positions[index] = selectedPositions.getOffset() + index;
+                }
+            }
+            if (page.trySelectPositions(positions, positionsOffset, selectedPositions.size())) {
+                selectedPositions = positionsRange(0, selectedPositions.size());
+            }
+        }
+
         return WorkProcessor.create(new ProjectSelectedPositions(session, memoryContext, metrics, page, selectedPositions));
     }
 
+    private static boolean isAllPositions(SelectedPositions selectedPositions, int positionCount)
+    {
+        return !selectedPositions.isList() && selectedPositions.getOffset() == 0 && selectedPositions.size() == positionCount;
+    }
+
     private class ProjectSelectedPositions
             implements WorkProcessor.Process<Page>
     {

core/trino-main/src/main/java/io/trino/operator/scalar/StringFunctions.java

@@ -964,7 +964,6 @@ public static Slice toUtf8(@LiteralParameter("x") long x, @SqlType("char(x)") Sl
         return Chars.padSpaces(slice, toIntExact(x));
     }
 
-    // TODO: implement N arguments char concat
     @Description("Concatenates given character strings")
     // Given CHAR type max length, if the result type is valid, allocation cannot fail.
     @ScalarFunction(neverFails = true)

core/trino-main/src/main/java/io/trino/sql/gen/columnar/ColumnarFilterEvaluator.java

@@ -38,8 +38,9 @@ public SelectionResult evaluate(ConnectorSession session, SelectedPositions acti
         if (activePositions.isEmpty()) {
             return new SelectionResult(activePositions, 0);
         }
-        // Should load only the blocks necessary for evaluating the kernel and unwrap lazy blocks
-        SourcePage loadedPage = filter.getInputChannels().getInputChannels(page);
+        // Materialize only the filter inputs here so reader code does not consume
+        // the generated filter's JIT inlining budget.
+        SourcePage loadedPage = SourcePage.create(filter.getInputChannels().getInputChannels(page).getPage());
         if (outputPositions.length < activePositions.size()) {
             outputPositions = new int[activePositions.size()];
         }

core/trino-main/src/test/java/io/trino/execution/TestQueryManagerConfig.java

@@ -77,6 +77,7 @@ public void testDefaults()
                 .setDispatcherQueryPoolSize(Integer.toString(max(50, Runtime.getRuntime().availableProcessors() * 10)))
                 .setQueryMaxScanPhysicalBytes(null)
                 .setQueryMaxWritePhysicalSize(null)
+                .setQueryMaxOutputDataSize(null)
                 .setRequiredWorkers(1)
                 .setRequiredWorkersMaxWait(new Duration(5, MINUTES))
                 .setRetryPolicy(RetryPolicy.NONE)
@@ -164,6 +165,7 @@ public void testExplicitPropertyMappings()
                 .put("query.dispatcher-query-pool-size", "151")
                 .put("query.max-scan-physical-bytes", "1kB")
                 .put("query.max-write-physical-size", "1TB")
+                .put("query.max-output-data-size", "1TB")
                 .put("query-manager.required-workers", "333")
                 .put("query-manager.required-workers-max-wait", "33m")
                 .put("retry-policy", "QUERY")
@@ -248,6 +250,7 @@ public void testExplicitPropertyMappings()
                 .setDispatcherQueryPoolSize("151")
                 .setQueryMaxScanPhysicalBytes(DataSize.of(1, KILOBYTE))
                 .setQueryMaxWritePhysicalSize(DataSize.of(1, TERABYTE))
+                .setQueryMaxOutputDataSize(DataSize.of(1, TERABYTE))
                 .setRequiredWorkers(333)
                 .setRequiredWorkersMaxWait(new Duration(33, MINUTES))
                 .setRetryPolicy(RetryPolicy.QUERY)