Merge branch 'prestodb:master' into cve_alluxio_shaded

Jenkinsfile

@@ -1,9 +1,50 @@
+AGENT_MAVEN = '''
+    apiVersion: v1
+    kind: Pod
+    spec:
+        containers:
+        - name: maven
+          image: maven:3.8.6-openjdk-8-slim
+          env:
+          - name: MAVEN_OPTS
+            value: "-Xmx8000m -Xms8000m"
+          resources:
+            requests:
+              memory: "16Gi"
+              cpu: "4000m"
+            limits:
+              memory: "16Gi"
+              cpu: "4000m"
+          tty: true
+          command:
+          - cat
+'''
+
+AGENT_DIND = '''
+    apiVersion: v1
+    kind: Pod
+    spec:
+        containers:
+        - name: dind
+          image: docker:20.10.16-dind-alpine3.15
+          securityContext:
+            privileged: true
+          tty: true
+          resources:
+            requests:
+              memory: "29Gi"
+              cpu: "7500m"
+            limits:
+              memory: "29Gi"
+              cpu: "7500m"
+'''
+
 pipeline {
 
     agent none
 
     environment {
-        AWS_CREDENTIAL_ID  = 'aws-jenkins'
+        AWS_CREDENTIAL_ID  = 'oss-presto-aws'
         AWS_DEFAULT_REGION = 'us-east-1'
         AWS_ECR            = 'public.ecr.aws/oss-presto'
         AWS_S3_PREFIX      = 's3://oss-prestodb/presto'
@@ -33,7 +74,7 @@ pipeline {
             agent {
                 kubernetes {
                     defaultContainer 'maven'
-                    yamlFile 'jenkins/agent-maven.yaml'
+                    yaml AGENT_MAVEN
                 }
             }
 
@@ -132,7 +173,7 @@ pipeline {
             agent {
                 kubernetes {
                     defaultContainer 'dind'
-                    yamlFile 'jenkins/agent-dind.yaml'
+                    yaml AGENT_DIND
                 }
             }
 

presto-docs/src/main/sphinx/admin/properties-session.rst

@@ -81,6 +81,18 @@ Number of local parallel table writer threads per worker for partitioned writes.
 set, the number set by ``task_writer_count`` will be used. It is required to be a power
 of two for a Java query engine.
 
+``single_node_execution_enabled``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+This property ensures that queries scheduled in this cluster use only a single
+node for execution, which may improve performance for small queries which can
+be executed within a single node.
+
+The corresponding configuration property is :ref:`admin/properties:\`\`single-node-execution-enabled\`\``.
+
 Spilling Properties
 -------------------
 

presto-docs/src/main/sphinx/admin/properties.rst

@@ -70,6 +70,18 @@ When enabled, the logical plan will begin to be built and validated before
 queueing and allocation of cluster resources so that any errors or
 incompatibilities in the query plan will fail quickly and inform the user.
 
+``single-node-execution-enabled``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+This property ensures that queries scheduled in this cluster use only a single
+node for execution, which may improve performance for small queries which can
+be executed within a single node.
+
+The corresponding session property is :ref:`admin/properties-session:\`\`single_node_execution_enabled\`\``.
+
 .. _tuning-memory:
 
 Memory Management Properties

presto-docs/src/main/sphinx/presto_cpp/properties-session.rst

@@ -307,11 +307,73 @@ Minimum number of rows to use prefix-sort.
 The default value has been derived using micro-benchmarking.
 
 ``native_op_trace_directory_create_config``
-^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
 * **Type:** ``varchar``
 * **Default value:** ``""``
 
 Native Execution only. Config used to create operator trace directory. This config is provided
 to underlying file system and the config is free form. The form should be defined by the
-underlying file system.
+underlying file system.
+
+``native_query_trace_enabled``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``boolean``
+* **Default value:** ``false``
+
+Enable query tracing. After enabled, trace data will be generated with query execution, and
+can be used by TraceReplayer. It needs to be used together with native_query_trace_node_ids,
+native_query_trace_max_bytes, native_query_trace_fragment_id, and native_query_trace_shard_id
+to match the task to be traced.
+
+
+``native_query_trace_dir``
+^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``varchar``
+* **Default value:** ``""``
+
+The location to store the trace files.
+
+``native_query_trace_node_ids``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``varchar``
+* **Default value:** ``""``
+
+A comma-separated list of plan node ids whose input data will be traced.
+Empty string if only want to trace the query metadata.
+
+``native_query_trace_max_bytes``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``integer``
+* **Default value:** ``0``
+
+The max trace bytes limit. Tracing is disabled if zero.
+
+``native_query_trace_fragment_id``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``varchar``
+* **Default value:** ``.*``
+
+The fragment id to be traced. If not specified, all fragments will be matched.
+
+``native_query_trace_shard_id``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``varchar``
+* **Default value:** ``.*``
+
+The shard id to be traced. If not specified, all shards will be matched.
+
+``native_query_trace_task_reg_exp``
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+* **Type:** ``varchar``
+* **Default value:** ``""``
+
+The regular expression to match a task for tracing. It will be deprecated if there is
+no issue with native_query_trace_fragment_id and native_query_trace_shard_id.

presto-hive/src/test/java/com/facebook/presto/hive/TestDistributedQueriesSingleNode.java

@@ -0,0 +1,59 @@
+/*
+ * 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 com.facebook.presto.hive;
+
+import com.facebook.presto.testing.QueryRunner;
+import com.facebook.presto.tests.AbstractTestDistributedQueries;
+import com.google.common.collect.ImmutableMap;
+import org.testng.annotations.Test;
+
+import java.util.Optional;
+
+import static io.airlift.tpch.TpchTable.getTables;
+
+@Test(singleThreaded = true)
+public class TestDistributedQueriesSingleNode
+        extends AbstractTestDistributedQueries
+{
+    @Override
+    protected QueryRunner createQueryRunner()
+            throws Exception
+    {
+        ImmutableMap.Builder<String, String> coordinatorProperties = ImmutableMap.builder();
+        coordinatorProperties.put("single-node-execution-enabled", "true");
+        return HiveQueryRunner.createQueryRunner(
+                getTables(),
+                ImmutableMap.of(),
+                coordinatorProperties.build(),
+                Optional.empty());
+    }
+
+    @Override
+    protected boolean supportsNotNullColumns()
+    {
+        return false;
+    }
+
+    @Override
+    public void testDelete()
+    {
+        // Hive connector currently does not support row-by-row delete
+    }
+
+    @Override
+    public void testUpdate()
+    {
+        // Updates are not supported by the connector
+    }
+}

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergNativeMetadata.java

@@ -59,6 +59,7 @@
 import static com.facebook.presto.iceberg.IcebergSessionProperties.getCompressionCodec;
 import static com.facebook.presto.iceberg.IcebergTableProperties.getFileFormat;
 import static com.facebook.presto.iceberg.IcebergTableProperties.getPartitioning;
+import static com.facebook.presto.iceberg.IcebergTableProperties.getTableLocation;
 import static com.facebook.presto.iceberg.IcebergTableType.DATA;
 import static com.facebook.presto.iceberg.IcebergUtil.VIEW_OWNER;
 import static com.facebook.presto.iceberg.IcebergUtil.createIcebergViewProperties;
@@ -76,6 +77,7 @@
 import static com.facebook.presto.iceberg.util.IcebergPrestoModelConverters.toPrestoSchemaTableName;
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
 import static com.facebook.presto.spi.StandardErrorCode.SCHEMA_NOT_EMPTY;
+import static com.google.common.base.Strings.isNullOrEmpty;
 import static com.google.common.base.Verify.verify;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
@@ -313,11 +315,23 @@ public ConnectorOutputTableHandle beginCreateTable(ConnectorSession session, Con
         FileFormat fileFormat = getFileFormat(tableMetadata.getProperties());
 
         try {
-            transaction = catalogFactory.getCatalog(session).newCreateTableTransaction(
-                    toIcebergTableIdentifier(schemaTableName, catalogFactory.isNestedNamespaceEnabled()),
-                    schema,
-                    partitionSpec,
-                    populateTableProperties(tableMetadata, fileFormat, session));
+            TableIdentifier tableIdentifier = toIcebergTableIdentifier(schemaTableName, catalogFactory.isNestedNamespaceEnabled());
+            String targetPath = getTableLocation(tableMetadata.getProperties());
+            if (!isNullOrEmpty(targetPath)) {
+                transaction = catalogFactory.getCatalog(session).newCreateTableTransaction(
+                        tableIdentifier,
+                        schema,
+                        partitionSpec,
+                        targetPath,
+                        populateTableProperties(tableMetadata, fileFormat, session));
+            }
+            else {
+                transaction = catalogFactory.getCatalog(session).newCreateTableTransaction(
+                        tableIdentifier,
+                        schema,
+                        partitionSpec,
+                        populateTableProperties(tableMetadata, fileFormat, session));
+            }
         }
         catch (AlreadyExistsException e) {
             throw new TableAlreadyExistsException(schemaTableName);