Merge master

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockTransferor.java

@@ -183,12 +183,12 @@ private void transferAllOutstanding() {
       if (numRetries > 0) {
         logger.error("Exception while beginning {} of {} outstanding blocks (after {} retries)", e,
           MDC.of(LogKeys.TRANSFER_TYPE$.MODULE$, listener.getTransferType()),
-          MDC.of(LogKeys.NUM_BLOCK_IDS$.MODULE$, blockIdsToTransfer.length),
+          MDC.of(LogKeys.NUM_BLOCKS$.MODULE$, blockIdsToTransfer.length),
           MDC.of(LogKeys.NUM_RETRY$.MODULE$, numRetries));
       } else {
         logger.error("Exception while beginning {} of {} outstanding blocks", e,
           MDC.of(LogKeys.TRANSFER_TYPE$.MODULE$, listener.getTransferType()),
-          MDC.of(LogKeys.NUM_BLOCK_IDS$.MODULE$, blockIdsToTransfer.length));
+          MDC.of(LogKeys.NUM_BLOCKS$.MODULE$, blockIdsToTransfer.length));
       }
       if (shouldRetry(e) && initiateRetry(e)) {
         // successfully initiated a retry
@@ -219,7 +219,7 @@ synchronized boolean initiateRetry(Throwable e) {
       MDC.of(LogKeys.TRANSFER_TYPE$.MODULE$, listener.getTransferType()),
       MDC.of(LogKeys.NUM_RETRY$.MODULE$, retryCount),
       MDC.of(LogKeys.MAX_ATTEMPTS$.MODULE$, maxRetries),
-      MDC.of(LogKeys.NUM_BLOCK_IDS$.MODULE$, outstandingBlocksIds.size()),
+      MDC.of(LogKeys.NUM_BLOCKS$.MODULE$, outstandingBlocksIds.size()),
       MDC.of(LogKeys.RETRY_WAIT_TIME$.MODULE$, retryWaitTime));
 
     try {

common/utils/src/main/resources/error/error-conditions.json

@@ -4731,6 +4731,12 @@
           "<treeNode>"
         ]
       },
+      "UNSUPPORTED_CORRELATED_EXPRESSION_IN_JOIN_CONDITION" : {
+        "message" : [
+          "Correlated subqueries in the join predicate cannot reference both join inputs:",
+          "<subqueryExpression>"
+        ]
+      },
       "UNSUPPORTED_CORRELATED_REFERENCE_DATA_TYPE" : {
         "message" : [
           "Correlated column reference '<expr>' cannot be <dataType> type."

common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala

@@ -125,10 +125,10 @@ object LogKeys {
   case object CONTAINER_ID extends LogKey
   case object CONTAINER_STATE extends LogKey
   case object CONTEXT extends LogKey
-  case object CONTEXT_CREATION_SITE extends LogKey
   case object COST extends LogKey
   case object COUNT extends LogKey
   case object CREATED_POOL_NAME extends LogKey
+  case object CREATION_SITE extends LogKey
   case object CREDENTIALS_RENEWAL_INTERVAL_RATIO extends LogKey
   case object CROSS_VALIDATION_METRIC extends LogKey
   case object CROSS_VALIDATION_METRICS extends LogKey
@@ -139,8 +139,9 @@ object LogKeys {
   case object CSV_SCHEMA_FIELD_NAMES extends LogKey
   case object CSV_SOURCE extends LogKey
   case object CURRENT_BATCH_ID extends LogKey
+  case object CURRENT_DISK_SIZE extends LogKey
   case object CURRENT_FILE extends LogKey
-  case object CURRENT_MEMORY_BYTES extends LogKey
+  case object CURRENT_MEMORY_SIZE extends LogKey
   case object CURRENT_PATH extends LogKey
   case object CURRENT_TIME extends LogKey
   case object DATA extends LogKey
@@ -155,7 +156,6 @@ object LogKeys {
   case object DEFAULT_COMPACT_INTERVAL extends LogKey
   case object DEFAULT_ISOLATION_LEVEL extends LogKey
   case object DEFAULT_NAME extends LogKey
-  case object DEFAULT_SCHEDULING_MODE extends LogKey
   case object DEFAULT_VALUE extends LogKey
   case object DELAY extends LogKey
   case object DELEGATE extends LogKey
@@ -216,6 +216,8 @@ object LogKeys {
   case object EXPR extends LogKey
   case object EXPR_TERMS extends LogKey
   case object EXTENDED_EXPLAIN_GENERATOR extends LogKey
+  case object FAILED_STAGE extends LogKey
+  case object FAILED_STAGE_NAME extends LogKey
   case object FAILURES extends LogKey
   case object FALLBACK_VERSION extends LogKey
   case object FEATURE_COLUMN extends LogKey
@@ -240,6 +242,7 @@ object LogKeys {
   case object FINAL_OUTPUT_PATH extends LogKey
   case object FINAL_PATH extends LogKey
   case object FINISH_TRIGGER_DURATION extends LogKey
+  case object FREE_MEMORY_SIZE extends LogKey
   case object FROM_OFFSET extends LogKey
   case object FROM_TIME extends LogKey
   case object FUNCTION_NAME extends LogKey
@@ -261,6 +264,7 @@ object LogKeys {
   case object HIVE_OPERATION_STATE extends LogKey
   case object HIVE_OPERATION_TYPE extends LogKey
   case object HOST extends LogKey
+  case object HOST_LOCAL_BLOCKS_SIZE extends LogKey
   case object HOST_NAME extends LogKey
   case object HOST_NAMES extends LogKey
   case object HOST_PORT extends LogKey
@@ -279,6 +283,7 @@ object LogKeys {
   case object INITIAL_REGISTRATION_RETRIES extends LogKey
   case object INIT_MODE extends LogKey
   case object INPUT extends LogKey
+  case object INPUT_SPLIT extends LogKey
   case object INTERVAL extends LogKey
   case object ISOLATION_LEVEL extends LogKey
   case object ISSUE_DATE extends LogKey
@@ -314,6 +319,7 @@ object LogKeys {
   case object LOAD_FACTOR extends LogKey
   case object LOAD_TIME extends LogKey
   case object LOCALE extends LogKey
+  case object LOCAL_BLOCKS_SIZE extends LogKey
   case object LOCAL_SCRATCH_DIR extends LogKey
   case object LOCATION extends LogKey
   case object LOGICAL_PLAN_COLUMNS extends LogKey
@@ -371,6 +377,7 @@ object LogKeys {
   case object MIN_SIZE extends LogKey
   case object MIN_TIME extends LogKey
   case object MIN_VERSION_NUM extends LogKey
+  case object MISSING_PARENT_STAGES extends LogKey
   case object MODEL_WEIGHTS extends LogKey
   case object MODULE_NAME extends LogKey
   case object NAMESPACE extends LogKey
@@ -389,8 +396,9 @@ object LogKeys {
   case object NORM extends LogKey
   case object NUM_ADDED_PARTITIONS extends LogKey
   case object NUM_APPS extends LogKey
+  case object NUM_ATTEMPT extends LogKey
   case object NUM_BIN extends LogKey
-  case object NUM_BLOCK_IDS extends LogKey
+  case object NUM_BLOCKS extends LogKey
   case object NUM_BROADCAST_BLOCK extends LogKey
   case object NUM_BYTES extends LogKey
   case object NUM_BYTES_CURRENT extends LogKey
@@ -409,13 +417,16 @@ object LogKeys {
   case object NUM_CORES extends LogKey
   case object NUM_DATA_FILE extends LogKey
   case object NUM_DATA_FILES extends LogKey
+  case object NUM_DECOMMISSIONED extends LogKey
   case object NUM_DRIVERS extends LogKey
   case object NUM_DROPPED_PARTITIONS extends LogKey
   case object NUM_EFFECTIVE_RULE_OF_RUNS extends LogKey
   case object NUM_ELEMENTS_SPILL_THRESHOLD extends LogKey
   case object NUM_EVENTS extends LogKey
   case object NUM_EXAMPLES extends LogKey
   case object NUM_EXECUTORS extends LogKey
+  case object NUM_EXECUTORS_EXITED extends LogKey
+  case object NUM_EXECUTORS_KILLED extends LogKey
   case object NUM_EXECUTOR_CORES extends LogKey
   case object NUM_EXECUTOR_CORES_REMAINING extends LogKey
   case object NUM_EXECUTOR_CORES_TOTAL extends LogKey
@@ -430,6 +441,7 @@ object LogKeys {
   case object NUM_FILES_REUSED extends LogKey
   case object NUM_FREQUENT_ITEMS extends LogKey
   case object NUM_GRACEFULLY_DECOMMISSIONED extends LogKey
+  case object NUM_HOST_LOCAL_BLOCKS extends LogKey
   case object NUM_INDEX_FILE extends LogKey
   case object NUM_INDEX_FILES extends LogKey
   case object NUM_ITERATIONS extends LogKey
@@ -438,6 +450,7 @@ object LogKeys {
   case object NUM_LEADING_SINGULAR_VALUES extends LogKey
   case object NUM_LEFT_PARTITION_VALUES extends LogKey
   case object NUM_LOADED_ENTRIES extends LogKey
+  case object NUM_LOCAL_BLOCKS extends LogKey
   case object NUM_LOCAL_DIRS extends LogKey
   case object NUM_LOCAL_FREQUENT_PATTERN extends LogKey
   case object NUM_MERGERS extends LogKey
@@ -457,8 +470,11 @@ object LogKeys {
   case object NUM_POINT extends LogKey
   case object NUM_PREFIXES extends LogKey
   case object NUM_PRUNED extends LogKey
+  case object NUM_PUSH_MERGED_LOCAL_BLOCKS extends LogKey
   case object NUM_RECORDS_READ extends LogKey
   case object NUM_RELEASED_LOCKS extends LogKey
+  case object NUM_REMAINED extends LogKey
+  case object NUM_REMOTE_BLOCKS extends LogKey
   case object NUM_REMOVED_WORKERS extends LogKey
   case object NUM_REPLICAS extends LogKey
   case object NUM_REQUESTS extends LogKey
@@ -474,13 +490,14 @@ object LogKeys {
   case object NUM_SPILL_INFOS extends LogKey
   case object NUM_SPILL_WRITERS extends LogKey
   case object NUM_SUB_DIRS extends LogKey
+  case object NUM_SUCCESSFUL_TASKS extends LogKey
   case object NUM_TASKS extends LogKey
   case object NUM_TASK_CPUS extends LogKey
   case object NUM_TRAIN_WORD extends LogKey
-  case object NUM_UNEXPECTEDLY_EXIT extends LogKey
-  case object NUM_UNFINISHED_DECOMMISSION extends LogKey
+  case object NUM_UNFINISHED_DECOMMISSIONED extends LogKey
   case object NUM_VERSIONS_RETAIN extends LogKey
   case object NUM_WEIGHTED_EXAMPLES extends LogKey
+  case object NUM_WORKERS extends LogKey
   case object OBJECT_AGG_SORT_BASED_FALLBACK_THRESHOLD extends LogKey
   case object OBJECT_ID extends LogKey
   case object OFFSET extends LogKey
@@ -497,12 +514,15 @@ object LogKeys {
   case object OPTIONS extends LogKey
   case object OP_ID extends LogKey
   case object OP_TYPE extends LogKey
+  case object ORIGINAL_DISK_SIZE extends LogKey
+  case object ORIGINAL_MEMORY_SIZE extends LogKey
   case object OS_ARCH extends LogKey
   case object OS_NAME extends LogKey
   case object OS_VERSION extends LogKey
   case object OUTPUT extends LogKey
   case object OVERHEAD_MEMORY_SIZE extends LogKey
   case object PAGE_SIZE extends LogKey
+  case object PARENT_STAGES extends LogKey
   case object PARSE_MODE extends LogKey
   case object PARTITIONED_FILE_READER extends LogKey
   case object PARTITIONER extends LogKey
@@ -540,6 +560,7 @@ object LogKeys {
   case object PROTOCOL_VERSION extends LogKey
   case object PROVIDER extends LogKey
   case object PUSHED_FILTERS extends LogKey
+  case object PUSH_MERGED_LOCAL_BLOCKS_SIZE extends LogKey
   case object PVC_METADATA_NAME extends LogKey
   case object PYTHON_EXEC extends LogKey
   case object PYTHON_PACKAGES extends LogKey
@@ -574,13 +595,14 @@ object LogKeys {
   case object REDUCE_ID extends LogKey
   case object REGEX extends LogKey
   case object REGISTERED_EXECUTOR_FILE extends LogKey
-  case object REGISTER_MERGE_RESULT extends LogKey
+  case object REGISTER_MERGE_RESULTS extends LogKey
   case object RELATION_NAME extends LogKey
   case object RELATION_OUTPUT extends LogKey
   case object RELATIVE_TOLERANCE extends LogKey
   case object RELEASED_LOCKS extends LogKey
   case object REMAINING_PARTITIONS extends LogKey
   case object REMOTE_ADDRESS extends LogKey
+  case object REMOTE_BLOCKS_SIZE extends LogKey
   case object REMOVE_FROM_MASTER extends LogKey
   case object REPORT_DETAILS extends LogKey
   case object REQUESTER_SIZE extends LogKey
@@ -611,6 +633,7 @@ object LogKeys {
   case object RUN_ID extends LogKey
   case object SCALA_VERSION extends LogKey
   case object SCHEDULER_POOL_NAME extends LogKey
+  case object SCHEDULING_MODE extends LogKey
   case object SCHEMA extends LogKey
   case object SCHEMA2 extends LogKey
   case object SERVER_NAME extends LogKey
@@ -622,7 +645,6 @@ object LogKeys {
   case object SESSION_KEY extends LogKey
   case object SET_CLIENT_INFO_REQUEST extends LogKey
   case object SHARD_ID extends LogKey
-  case object SHELL_OPTIONS extends LogKey
   case object SHORT_USER_NAME extends LogKey
   case object SHUFFLE_BLOCK_INFO extends LogKey
   case object SHUFFLE_DB_BACKEND_KEY extends LogKey
@@ -654,21 +676,18 @@ object LogKeys {
   case object SQL_TEXT extends LogKey
   case object SRC_PATH extends LogKey
   case object STAGE extends LogKey
-  case object STAGE2 extends LogKey
   case object STAGES extends LogKey
   case object STAGE_ATTEMPT extends LogKey
   case object STAGE_ATTEMPT_ID extends LogKey
   case object STAGE_ATTEMPT_NUMBER extends LogKey
   case object STAGE_ID extends LogKey
   case object STAGE_NAME extends LogKey
-  case object STAGE_NAME2 extends LogKey
-  case object STAGE_PARENTS extends LogKey
-  case object STAGE_STATUS extends LogKey
   case object START_INDEX extends LogKey
   case object STATEMENT_ID extends LogKey
   case object STATE_STORE_ID extends LogKey
   case object STATE_STORE_PROVIDER extends LogKey
   case object STATE_STORE_VERSION extends LogKey
+  case object STATS extends LogKey
   case object STATUS extends LogKey
   case object STAT_COUNTER extends LogKey
   case object STDERR extends LogKey
@@ -695,16 +714,20 @@ object LogKeys {
   case object TABLE_NAME extends LogKey
   case object TABLE_TYPE extends LogKey
   case object TABLE_TYPES extends LogKey
+  case object TAG extends LogKey
   case object TARGET_NUM_EXECUTOR extends LogKey
   case object TARGET_NUM_EXECUTOR_DELTA extends LogKey
   case object TARGET_PATH extends LogKey
   case object TASK extends LogKey
   case object TASK_ATTEMPT_ID extends LogKey
   case object TASK_ID extends LogKey
+  case object TASK_LOCALITY extends LogKey
   case object TASK_NAME extends LogKey
   case object TASK_REQUIREMENTS extends LogKey
   case object TASK_RESOURCES extends LogKey
-  case object TASK_SET_MANAGER_NAME extends LogKey
+  case object TASK_RESOURCE_ASSIGNMENTS extends LogKey
+  case object TASK_SET_ID extends LogKey
+  case object TASK_SET_MANAGER extends LogKey
   case object TASK_SET_NAME extends LogKey
   case object TASK_STATE extends LogKey
   case object TEMP_FILE extends LogKey
@@ -738,6 +761,7 @@ object LogKeys {
   case object TOTAL_EFFECTIVE_TIME extends LogKey
   case object TOTAL_RECORDS_READ extends LogKey
   case object TOTAL_REGISTRATION_RETRIES extends LogKey
+  case object TOTAL_SIZE extends LogKey
   case object TOTAL_TIME extends LogKey
   case object TOTAL_TIME_READ extends LogKey
   case object TO_TIME extends LogKey

connector/connect/common/src/test/resources/query-tests/explain-results/function_shiftleft.explain

@@ -1,2 +1,2 @@
-Project [shiftleft(cast(b#0 as int), 2) AS shiftleft(b, 2)#0]
+Project [(cast(b#0 as int) << 2) AS (b << 2)#0]
 +- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

connector/connect/common/src/test/resources/query-tests/explain-results/function_shiftright.explain

@@ -1,2 +1,2 @@
-Project [shiftright(cast(b#0 as int), 2) AS shiftright(b, 2)#0]
+Project [(cast(b#0 as int) >> 2) AS (b >> 2)#0]
 +- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

connector/connect/common/src/test/resources/query-tests/explain-results/function_shiftrightunsigned.explain

@@ -1,2 +1,2 @@
-Project [shiftrightunsigned(cast(b#0 as int), 2) AS shiftrightunsigned(b, 2)#0]
+Project [(cast(b#0 as int) >>> 2) AS (b >>> 2)#0]
 +- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

connector/connect/common/src/test/resources/query-tests/explain-results/grouping_and_grouping_id.explain

@@ -1,4 +1,4 @@
-Aggregate [a#0, b#0, spark_grouping_id#0L], [a#0, b#0, cast((shiftright(spark_grouping_id#0L, 1) & 1) as tinyint) AS grouping(a)#0, cast((shiftright(spark_grouping_id#0L, 0) & 1) as tinyint) AS grouping(b)#0, spark_grouping_id#0L AS grouping_id(a, b)#0L]
+Aggregate [a#0, b#0, spark_grouping_id#0L], [a#0, b#0, cast(((spark_grouping_id#0L >> 1) & 1) as tinyint) AS grouping(a)#0, cast(((spark_grouping_id#0L >> 0) & 1) as tinyint) AS grouping(b)#0, spark_grouping_id#0L AS grouping_id(a, b)#0L]
 +- Expand [[id#0L, a#0, b#0, a#0, b#0, 0], [id#0L, a#0, b#0, a#0, null, 1], [id#0L, a#0, b#0, null, b#0, 2], [id#0L, a#0, b#0, null, null, 3]], [id#0L, a#0, b#0, a#0, b#0, spark_grouping_id#0L]
    +- Project [id#0L, a#0, b#0, a#0 AS a#0, b#0 AS b#0]
       +- LocalRelation <empty>, [id#0L, a#0, b#0]

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MariaDBKrbIntegrationSuite.scala

@@ -25,9 +25,9 @@ import org.apache.spark.sql.execution.datasources.jdbc.connection.SecureConnecti
 import org.apache.spark.tags.DockerTest
 
 /**
- * To run this test suite for a specific version (e.g., mariadb:10.5.12):
+ * To run this test suite for a specific version (e.g., mariadb:10.5.25):
  * {{{
- *   ENABLE_DOCKER_INTEGRATION_TESTS=1 MARIADB_DOCKER_IMAGE_NAME=mariadb:10.5.12
+ *   ENABLE_DOCKER_INTEGRATION_TESTS=1 MARIADB_DOCKER_IMAGE_NAME=mariadb:10.5.25
  *     ./build/sbt -Pdocker-integration-tests
  *     "docker-integration-tests/testOnly org.apache.spark.sql.jdbc.MariaDBKrbIntegrationSuite"
  * }}}
@@ -38,7 +38,7 @@ class MariaDBKrbIntegrationSuite extends DockerKrbJDBCIntegrationSuite {
   override protected val keytabFileName = "mariadb.keytab"
 
   override val db = new DatabaseOnDocker {
-    override val imageName = sys.env.getOrElse("MARIADB_DOCKER_IMAGE_NAME", "mariadb:10.5.12")
+    override val imageName = sys.env.getOrElse("MARIADB_DOCKER_IMAGE_NAME", "mariadb:10.5.25")
     override val env = Map(
       "MYSQL_ROOT_PASSWORD" -> "rootpass"
     )

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/MySQLDatabaseOnDocker.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.jdbc
 
 class MySQLDatabaseOnDocker extends DatabaseOnDocker {
-  override val imageName = sys.env.getOrElse("MYSQL_DOCKER_IMAGE_NAME", "mysql:8.3.0")
+  override val imageName = sys.env.getOrElse("MYSQL_DOCKER_IMAGE_NAME", "mysql:8.4.0")
   override val env = Map(
     "MYSQL_ROOT_PASSWORD" -> "rootpass"
   )

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresIntegrationSuite.scala

@@ -32,17 +32,17 @@ import org.apache.spark.sql.types._
 import org.apache.spark.tags.DockerTest
 
 /**
- * To run this test suite for a specific version (e.g., postgres:16.2):
+ * To run this test suite for a specific version (e.g., postgres:16.3-alpine):
  * {{{
- *   ENABLE_DOCKER_INTEGRATION_TESTS=1 POSTGRES_DOCKER_IMAGE_NAME=postgres:16.2
+ *   ENABLE_DOCKER_INTEGRATION_TESTS=1 POSTGRES_DOCKER_IMAGE_NAME=postgres:16.3-alpine
  *     ./build/sbt -Pdocker-integration-tests
  *     "docker-integration-tests/testOnly org.apache.spark.sql.jdbc.PostgresIntegrationSuite"
  * }}}
  */
 @DockerTest
 class PostgresIntegrationSuite extends DockerJDBCIntegrationSuite {
   override val db = new DatabaseOnDocker {
-    override val imageName = sys.env.getOrElse("POSTGRES_DOCKER_IMAGE_NAME", "postgres:16.2-alpine")
+    override val imageName = sys.env.getOrElse("POSTGRES_DOCKER_IMAGE_NAME", "postgres:16.3-alpine")
     override val env = Map(
       "POSTGRES_PASSWORD" -> "rootpass"
     )

connector/docker-integration-tests/src/test/scala/org/apache/spark/sql/jdbc/PostgresKrbIntegrationSuite.scala

@@ -25,9 +25,9 @@ import org.apache.spark.sql.execution.datasources.jdbc.connection.SecureConnecti
 import org.apache.spark.tags.DockerTest
 
 /**
- * To run this test suite for a specific version (e.g., postgres:16.2):
+ * To run this test suite for a specific version (e.g., postgres:16.3-alpine):
  * {{{
- *   ENABLE_DOCKER_INTEGRATION_TESTS=1 POSTGRES_DOCKER_IMAGE_NAME=postgres:16.2
+ *   ENABLE_DOCKER_INTEGRATION_TESTS=1 POSTGRES_DOCKER_IMAGE_NAME=postgres:16.3-alpine
  *     ./build/sbt -Pdocker-integration-tests
  *     "docker-integration-tests/testOnly *PostgresKrbIntegrationSuite"
  * }}}
@@ -38,7 +38,7 @@ class PostgresKrbIntegrationSuite extends DockerKrbJDBCIntegrationSuite {
   override protected val keytabFileName = "postgres.keytab"
 
   override val db = new DatabaseOnDocker {
-    override val imageName = sys.env.getOrElse("POSTGRES_DOCKER_IMAGE_NAME", "postgres:16.2")
+    override val imageName = sys.env.getOrElse("POSTGRES_DOCKER_IMAGE_NAME", "postgres:16.3-alpine")
     override val env = Map(
       "POSTGRES_PASSWORD" -> "rootpass"
     )