diff --git a/Jenkinsfile b/Jenkinsfile
new file mode 100644
index 0000000000000..29cb7d2b7d176
--- /dev/null
+++ b/Jenkinsfile
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+pipeline {
+  agent { label 'local' }
+
+  options {
+    disableConcurrentBuilds()
+    skipDefaultCheckout(true)
+    timestamps()
+    buildDiscarder(logRotator(numToKeepStr: '20'))
+  }
+
+  parameters {
+    string(name: 'BRANCH_TO_BUILD', defaultValue: 'branch-4.1', description: 'Git branch to build.')
+    booleanParam(name: 'FORCE_UPDATE', defaultValue: true, description: 'Pass -U to Maven to refresh snapshot and cached dependency resolution.')
+  }
+
+  environment {
+    BUILD_PROFILES = 'kubernetes,hadoop-provided,parquet-provided,hive,hadoop-cloud'
+    DISTRIBUTION_PROFILES = 'kubernetes,hadoop-provided,parquet-provided,hive,hadoop-cloud,bigtop-dist'
+    DOCKER_IMAGE = 'maven:3.9.9-eclipse-temurin-17'
+    HOST_MAVEN_REPO = '/home/jenkinsmaster/.m2'
+    DISTRIBUTION_REPOSITORY = '/opt/repository/master'
+    SPARK_SQL_DEPS_REPOSITORY = '/opt/repository/master/spark-sql-dependencies/spark-4.1'
+    MAVEN_LOCAL_REPO = '/maven-repo/repository'
+    MAVEN_OPTS = '-Xmx4G'
+    MAVEN_SETTINGS = "${WORKSPACE}@tmp/mvn-settings.xml"
+  }
+
+  stages {
+    stage('Checkout') {
+      steps {
+        deleteDir()
+        checkout([$class: 'GitSCM',
+          branches: [[name: "${params.BRANCH_TO_BUILD}"]],
+          userRemoteConfigs: [[
+            url: 'git@github.com:gibchikafa/spark.git',
+            credentialsId: 'id_rsa'
+          ]]
+        ])
+      }
+    }
+
+    stage('Prepare Maven') {
+      steps {
+        withCredentials([usernamePassword(credentialsId: 'a0770738-4ef3-4acc-a6ba-097ee6c85b44', passwordVariable: 'PASSWORD', usernameVariable: 'USERNAME')]) {
+          sh '''#!/bin/bash -eu
+            rm -rf "$WORKSPACE/.m2" "$HOST_MAVEN_REPO/repository/org/apache/spark"
+            mkdir -p "$(dirname "$MAVEN_SETTINGS")" "$HOST_MAVEN_REPO/repository"
+            cat > "$MAVEN_SETTINGS" <<EOF
+<settings>
+  <localRepository>${MAVEN_LOCAL_REPO}</localRepository>
+  <servers>
+    <server>
+      <id>HopsEE</id>
+      <username>$USERNAME</username>
+      <password>$PASSWORD</password>
+    </server>
+    <server>
+      <id>Hops</id>
+      <username>$USERNAME</username>
+      <password>$PASSWORD</password>
+    </server>
+    <server>
+      <id>hops-releases</id>
+      <username>$USERNAME</username>
+      <password>$PASSWORD</password>
+    </server>
+    <server>
+      <id>Spark</id>
+      <username>$USERNAME</username>
+      <password>$PASSWORD</password>
+    </server>
+  </servers>
+</settings>
+EOF
+          '''
+        }
+      }
+    }
+
+    stage('Resolve Version') {
+      steps {
+        sh '''#!/bin/bash -eu
+          perl -0ne 'if (m{<artifactId>spark-parent_2\\.13</artifactId>\\s*<version>([^<]+)</version>}) { print "$1"; exit }' pom.xml > version.log
+          echo "POM_VERSION=$(cat version.log)"
+        '''
+      }
+    }
+
+    stage('Build and Deploy') {
+      steps {
+        sh '''#!/bin/bash -eu
+          UPDATE_ARG=""
+          if [ "$FORCE_UPDATE" = "true" ]; then
+            UPDATE_ARG="-U"
+          fi
+
+          docker run --rm \
+            -u "$(id -u):$(id -g)" \
+            -v "$WORKSPACE:$WORKSPACE" \
+            -v "$(dirname "$MAVEN_SETTINGS"):$(dirname "$MAVEN_SETTINGS")" \
+            -v "$HOST_MAVEN_REPO:/maven-repo" \
+            -w "$WORKSPACE" \
+            -e GITHUB_ACTIONS=true \
+            -e HOME=/tmp \
+            -e MAVEN_CONFIG=/tmp/maven-config \
+            -e MAVEN_LOCAL_REPO="$MAVEN_LOCAL_REPO" \
+            -e MAVEN_OPTS="$MAVEN_OPTS" \
+            -e MAVEN_SETTINGS="$MAVEN_SETTINGS" \
+            -e BUILD_PROFILES="$BUILD_PROFILES" \
+            -e UPDATE_ARG="$UPDATE_ARG" \
+            "$DOCKER_IMAGE" \
+            bash -lc '
+              set -eu
+              export PATH="$JAVA_HOME/bin:$PATH"
+              JAVA_VERSION="$("$JAVA_HOME/bin/java" -XshowSettings:properties -version 2>&1 | awk -F"= " "/java.specification.version =/{print \\$2; exit}")"
+              if [ "$JAVA_VERSION" != "17" ]; then
+                echo "Java 17 is required, but JAVA_HOME=$JAVA_HOME reports java.specification.version=$JAVA_VERSION" >&2
+                exit 1
+              fi
+              test -x "$JAVA_HOME/bin/javadoc"
+              ./build/mvn -s "$MAVEN_SETTINGS" -Dmaven.repo.local="$MAVEN_LOCAL_REPO" $UPDATE_ARG \
+                clean deploy -DskipTests -P"$BUILD_PROFILES"
+            '
+        '''
+      }
+    }
+
+    stage('Publish Spark SQL Dependencies') {
+      steps {
+        sh '''#!/bin/bash -eu
+          SPARK_VERSION="$(cat version.log)"
+          mkdir -p "$SPARK_SQL_DEPS_REPOSITORY"
+
+          cp "sql/core/target/spark-sql_2.13-${SPARK_VERSION}.jar" "$SPARK_SQL_DEPS_REPOSITORY/"
+          cp "connector/kafka-0-10-sql/target/spark-sql-kafka-0-10_2.13-${SPARK_VERSION}.jar" "$SPARK_SQL_DEPS_REPOSITORY/"
+          cp "connector/avro/target/spark-avro_2.13-${SPARK_VERSION}.jar" "$SPARK_SQL_DEPS_REPOSITORY/"
+
+          ls -l \
+            "$SPARK_SQL_DEPS_REPOSITORY/spark-sql_2.13-${SPARK_VERSION}.jar" \
+            "$SPARK_SQL_DEPS_REPOSITORY/spark-sql-kafka-0-10_2.13-${SPARK_VERSION}.jar" \
+            "$SPARK_SQL_DEPS_REPOSITORY/spark-avro_2.13-${SPARK_VERSION}.jar"
+        '''
+      }
+    }
+
+    stage('Build Distribution') {
+      steps {
+        sh '''#!/bin/bash -eu
+          UPDATE_ARG=""
+          if [ "$FORCE_UPDATE" = "true" ]; then
+            UPDATE_ARG="-U"
+          fi
+
+          docker run --rm \
+            -u "$(id -u):$(id -g)" \
+            -v "$WORKSPACE:$WORKSPACE" \
+            -v "$(dirname "$MAVEN_SETTINGS"):$(dirname "$MAVEN_SETTINGS")" \
+            -v "$HOST_MAVEN_REPO:/maven-repo" \
+            -w "$WORKSPACE" \
+            -e GITHUB_ACTIONS=true \
+            -e HOME=/tmp \
+            -e MAVEN_CONFIG=/tmp/maven-config \
+            -e MAVEN_LOCAL_REPO="$MAVEN_LOCAL_REPO" \
+            -e MAVEN_OPTS="$MAVEN_OPTS" \
+            -e MAVEN_SETTINGS="$MAVEN_SETTINGS" \
+            -e DISTRIBUTION_PROFILES="$DISTRIBUTION_PROFILES" \
+            -e UPDATE_ARG="$UPDATE_ARG" \
+            "$DOCKER_IMAGE" \
+            bash -lc '
+              set -eu
+              export PATH="$JAVA_HOME/bin:$PATH"
+              JAVA_VERSION="$("$JAVA_HOME/bin/java" -XshowSettings:properties -version 2>&1 | awk -F"= " "/java.specification.version =/{print \\$2; exit}")"
+              if [ "$JAVA_VERSION" != "17" ]; then
+                echo "Java 17 is required, but JAVA_HOME=$JAVA_HOME reports java.specification.version=$JAVA_VERSION" >&2
+                exit 1
+              fi
+              ./dev/make-distribution.sh --name without-hadoop-with-hive --tgz \
+                -s "$MAVEN_SETTINGS" -Dmaven.repo.local="$MAVEN_LOCAL_REPO" $UPDATE_ARG \
+                -P"$DISTRIBUTION_PROFILES"
+            '
+
+          mkdir -p "$DISTRIBUTION_REPOSITORY"
+          cp spark-*-bin-without-hadoop-with-hive.tgz "$DISTRIBUTION_REPOSITORY/"
+          ls -l spark-*-bin-without-hadoop-with-hive.tgz "$DISTRIBUTION_REPOSITORY"/spark-*-bin-without-hadoop-with-hive.tgz
+        '''
+      }
+    }
+  }
+
+  post {
+    always {
+      sh '''#!/bin/bash
+        rm -f "$MAVEN_SETTINGS"
+      '''
+      archiveArtifacts artifacts: 'version.log,spark-*-bin-without-hadoop-with-hive.tgz', allowEmptyArchive: true
+    }
+  }
+}
diff --git a/assembly/pom.xml b/assembly/pom.xml
index 2b78c2e9d84da..c741cd651294f 100644
--- a/assembly/pom.xml
+++ b/assembly/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
diff --git a/common/kvstore/pom.xml b/common/kvstore/pom.xml
index 5dfb4feb210d1..6fe92dab29d96 100644
--- a/common/kvstore/pom.xml
+++ b/common/kvstore/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/common/network-common/pom.xml b/common/network-common/pom.xml
index 2924e81bdb401..e1c2ab04d2732 100644
--- a/common/network-common/pom.xml
+++ b/common/network-common/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/common/network-shuffle/pom.xml b/common/network-shuffle/pom.xml
index 9d7e00cad47f1..4ff5693e23a76 100644
--- a/common/network-shuffle/pom.xml
+++ b/common/network-shuffle/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/common/network-yarn/pom.xml b/common/network-yarn/pom.xml
index 614cbde686d08..91e8dbc88f4cc 100644
--- a/common/network-yarn/pom.xml
+++ b/common/network-yarn/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -64,12 +64,12 @@
 
     <!-- Provided dependencies -->
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-api</artifactId>
       <version>${hadoop.version}</version>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
       <version>${hadoop.version}</version>
     </dependency>
diff --git a/common/sketch/pom.xml b/common/sketch/pom.xml
index f7248017eea5c..cd1c41ed7a6cc 100644
--- a/common/sketch/pom.xml
+++ b/common/sketch/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/common/tags/pom.xml b/common/tags/pom.xml
index b53bbbdd9f195..196816571297b 100644
--- a/common/tags/pom.xml
+++ b/common/tags/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/common/unsafe/pom.xml b/common/unsafe/pom.xml
index ae566ab94ede0..0a7a6563fa5bb 100644
--- a/common/unsafe/pom.xml
+++ b/common/unsafe/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/common/utils-java/pom.xml b/common/utils-java/pom.xml
index c37008ea4a4a9..31c3de935558d 100644
--- a/common/utils-java/pom.xml
+++ b/common/utils-java/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/common/utils/pom.xml b/common/utils/pom.xml
index 9e1d2126bb856..e9dfbb46a4834 100644
--- a/common/utils/pom.xml
+++ b/common/utils/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/common/variant/pom.xml b/common/variant/pom.xml
index c9c97661a62c7..5241f33740c62 100644
--- a/common/variant/pom.xml
+++ b/common/variant/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/connector/avro/pom.xml b/connector/avro/pom.xml
index a6dc25a6d3a41..bb1d8780d1ba9 100644
--- a/connector/avro/pom.xml
+++ b/connector/avro/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/connector/docker-integration-tests/pom.xml b/connector/docker-integration-tests/pom.xml
index c0680351958ef..7c66813d0e8be 100644
--- a/connector/docker-integration-tests/pom.xml
+++ b/connector/docker-integration-tests/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -81,7 +81,7 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-minikdc</artifactId>
       <scope>test</scope>
     </dependency>
diff --git a/connector/kafka-0-10-assembly/pom.xml b/connector/kafka-0-10-assembly/pom.xml
index 85df48e6631c8..019bdbdc7af6f 100644
--- a/connector/kafka-0-10-assembly/pom.xml
+++ b/connector/kafka-0-10-assembly/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -70,13 +70,13 @@
       <scope>provided</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-api</artifactId>
       <version>${hadoop.version}</version>
       <scope>provided</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
       <version>${hadoop.version}</version>
       <scope>provided</scope>
diff --git a/connector/kafka-0-10-sql/pom.xml b/connector/kafka-0-10-sql/pom.xml
index 9b0a8282320b9..8626c5c72bc62 100644
--- a/connector/kafka-0-10-sql/pom.xml
+++ b/connector/kafka-0-10-sql/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -123,7 +123,7 @@
       </exclusions>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-minikdc</artifactId>
     </dependency>
     <dependency>
diff --git a/connector/kafka-0-10-token-provider/pom.xml b/connector/kafka-0-10-token-provider/pom.xml
index 9d5d563b07d73..e5d834fdbbc61 100644
--- a/connector/kafka-0-10-token-provider/pom.xml
+++ b/connector/kafka-0-10-token-provider/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -75,7 +75,7 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
       <scope>${hadoop.deps.scope}</scope>
     </dependency>
diff --git a/connector/kafka-0-10/pom.xml b/connector/kafka-0-10/pom.xml
index 0661c097a1807..89f5c50815deb 100644
--- a/connector/kafka-0-10/pom.xml
+++ b/connector/kafka-0-10/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/connector/kinesis-asl-assembly/pom.xml b/connector/kinesis-asl-assembly/pom.xml
index 5cb6ea8f144f1..59e4da5ae78cd 100644
--- a/connector/kinesis-asl-assembly/pom.xml
+++ b/connector/kinesis-asl-assembly/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -90,13 +90,13 @@
       <scope>provided</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-api</artifactId>
       <version>${hadoop.version}</version>
       <scope>provided</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
       <version>${hadoop.version}</version>
       <scope>provided</scope>
diff --git a/connector/kinesis-asl/pom.xml b/connector/kinesis-asl/pom.xml
index 340790c7b3131..f833dcb618bcd 100644
--- a/connector/kinesis-asl/pom.xml
+++ b/connector/kinesis-asl/pom.xml
@@ -20,7 +20,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/connector/profiler/pom.xml b/connector/profiler/pom.xml
index 1e6661acef795..51cf063b9728b 100644
--- a/connector/profiler/pom.xml
+++ b/connector/profiler/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/connector/protobuf/pom.xml b/connector/protobuf/pom.xml
index 13464b65cb315..138bec46320b9 100644
--- a/connector/protobuf/pom.xml
+++ b/connector/protobuf/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/connector/spark-ganglia-lgpl/pom.xml b/connector/spark-ganglia-lgpl/pom.xml
index f6f82ef349ba7..54e4d6d2f0893 100644
--- a/connector/spark-ganglia-lgpl/pom.xml
+++ b/connector/spark-ganglia-lgpl/pom.xml
@@ -20,7 +20,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/core/pom.xml b/core/pom.xml
index 000be688d71c9..b13955c16ba2f 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
@@ -73,14 +73,16 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-api</artifactId>
       <version>${hadoop.version}</version>
+      <scope>${hadoop.deps.scope}</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
       <version>${hadoop.version}</version>
+      <scope>${hadoop.deps.scope}</scope>
     </dependency>
     <dependency>
       <groupId>org.apache.spark</groupId>
@@ -420,7 +422,7 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-minikdc</artifactId>
       <scope>test</scope>
     </dependency>
@@ -480,13 +482,13 @@
     </dependency>
 
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-aws</artifactId>
       <version>${hadoop.version}</version>
       <scope>test</scope>
       <exclusions>
         <exclusion>
-          <groupId>org.apache.hadoop</groupId>
+          <groupId>${hadoop.group}</groupId>
           <artifactId>hadoop-common</artifactId>
         </exclusion>
         <exclusion>
diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
index 16ec7689d5261..65b95bcc977c7 100644
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
@@ -210,7 +210,9 @@ private[spark] class ExecutorAllocationManager(
     val shuffleDecommissionEnabled = decommissionEnabled &&
       conf.get(config.STORAGE_DECOMMISSION_SHUFFLE_BLOCKS_ENABLED)
     if (!conf.get(config.SHUFFLE_SERVICE_ENABLED) && !reliableShuffleStorage) {
-      if (shuffleTrackingEnabled) {
+      if (conf.isRssEnable()) {
+        logInfo("Dynamic allocation will use remote shuffle service")
+      } else if (shuffleTrackingEnabled) {
         logInfo("Dynamic allocation is enabled without a shuffle service.")
       } else if (shuffleDecommissionEnabled) {
         logInfo("Shuffle data decommission is enabled without a shuffle service.")
diff --git a/core/src/main/scala/org/apache/spark/SparkConf.scala b/core/src/main/scala/org/apache/spark/SparkConf.scala
index 70912bebe326d..88d36cf2432be 100644
--- a/core/src/main/scala/org/apache/spark/SparkConf.scala
+++ b/core/src/main/scala/org/apache/spark/SparkConf.scala
@@ -651,6 +651,11 @@ class SparkConf(loadDefaults: Boolean)
     Utils.redact(this, getAll).sorted.map { case (k, v) => k + "=" + v }.mkString("\n")
   }
 
+  def isRssEnable(): Boolean = {
+    val shuffleMgr = get("spark.shuffle.manager", "sort")
+    shuffleMgr.contains("RssShuffleManager") || shuffleMgr.contains("UniffleShuffleManager")
+  }
+
 }
 
 private[spark] object SparkConf extends Logging {
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
index d78d49335183a..f6aa64d5337c7 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
@@ -31,8 +31,9 @@ import scala.xml.Node
 
 import com.fasterxml.jackson.annotation.JsonIgnore
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize
-import org.apache.hadoop.fs.{FileStatus, FileSystem, Path, SafeModeAction}
+import org.apache.hadoop.fs.{FileStatus, FileSystem, Path}
 import org.apache.hadoop.hdfs.DistributedFileSystem
+import org.apache.hadoop.hdfs.protocol.HdfsConstants
 import org.apache.hadoop.security.AccessControlException
 
 import org.apache.spark.{SecurityManager, SparkConf, SparkException}
@@ -1237,7 +1238,7 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
 
   private[history] def isFsInSafeMode(dfs: DistributedFileSystem): Boolean = {
     /* true to check only for Active NNs status */
-    dfs.setSafeMode(SafeModeAction.GET, true)
+    dfs.setSafeMode(HdfsConstants.SafeModeAction.SAFEMODE_GET, true)
   }
 
   /**
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 69e766ebcef25..e335db53513cb 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -2887,7 +2887,8 @@ private[spark] class DAGScheduler(
     // we know to unregister shuffle output.  (Note that "worker" specifically refers to the process
     // from a Standalone cluster, where the shuffle service lives in the Worker.)
     val fileLost = !sc.shuffleDriverComponents.supportsReliableStorage() &&
-      (workerHost.isDefined || !env.blockManager.externalShuffleServiceEnabled)
+      (workerHost.isDefined || !env.blockManager.externalShuffleServiceEnabled) &&
+      !sc.getConf.isRssEnable()
     removeExecutorAndUnregisterOutputs(
       execId = execId,
       fileLost = fileLost,
diff --git a/core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala b/core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala
index 1e46142fab255..2907a3225ea1f 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala
@@ -154,96 +154,96 @@ private[spark] class EventLoggingListener(
     }
 
     // log stage completed event
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onJobStart(event: SparkListenerJobStart): Unit = {
-    logEvent(event.copy(properties = redactProperties(event.properties)), flushLogger = true)
+    logEvent(event.copy(properties = redactProperties(event.properties)))
   }
 
   override def onJobEnd(event: SparkListenerJobEnd): Unit = logEvent(event, flushLogger = true)
 
   override def onBlockManagerAdded(event: SparkListenerBlockManagerAdded): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onBlockManagerRemoved(event: SparkListenerBlockManagerRemoved): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onUnpersistRDD(event: SparkListenerUnpersistRDD): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onApplicationStart(event: SparkListenerApplicationStart): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onApplicationEnd(event: SparkListenerApplicationEnd): Unit = {
     logEvent(event, flushLogger = true)
   }
   override def onExecutorAdded(event: SparkListenerExecutorAdded): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onExecutorRemoved(event: SparkListenerExecutorRemoved): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onExecutorBlacklisted(event: SparkListenerExecutorBlacklisted): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onExecutorExcluded(event: SparkListenerExecutorExcluded): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onExecutorBlacklistedForStage(
       event: SparkListenerExecutorBlacklistedForStage): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onExecutorExcludedForStage(
       event: SparkListenerExecutorExcludedForStage): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onNodeBlacklistedForStage(event: SparkListenerNodeBlacklistedForStage): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onNodeExcludedForStage(event: SparkListenerNodeExcludedForStage): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onExecutorUnblacklisted(event: SparkListenerExecutorUnblacklisted): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onExecutorUnexcluded(event: SparkListenerExecutorUnexcluded): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
 
   override def onNodeBlacklisted(event: SparkListenerNodeBlacklisted): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onNodeExcluded(event: SparkListenerNodeExcluded): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onNodeUnblacklisted(event: SparkListenerNodeUnblacklisted): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onNodeUnexcluded(event: SparkListenerNodeUnexcluded): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onBlockUpdated(event: SparkListenerBlockUpdated): Unit = {
     if (shouldLogBlockUpdates) {
-      logEvent(event, flushLogger = true)
+      logEvent(event)
     }
   }
 
@@ -268,12 +268,12 @@ private[spark] class EventLoggingListener(
   }
 
   override def onResourceProfileAdded(event: SparkListenerResourceProfileAdded): Unit = {
-    logEvent(event, flushLogger = true)
+    logEvent(event)
   }
 
   override def onOtherEvent(event: SparkListenerEvent): Unit = {
     if (event.logEvent) {
-      logEvent(event, flushLogger = true)
+      logEvent(event)
     }
   }
 
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
index 69e0a10a34b28..7443353ce52b5 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
@@ -1146,7 +1146,7 @@ private[spark] class TaskSetManager(
     // could serve the shuffle outputs or the executor lost is caused by decommission (which
     // can destroy the whole host). The reason is the next stage wouldn't be able to fetch the
     // data from this dead executor so we would need to rerun these tasks on other executors.
-    val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+    val maybeShuffleMapOutputLoss = isShuffleMapTasks && !sched.sc.conf.isRssEnable() &&
       !sched.sc.shuffleDriverComponents.supportsReliableStorage() &&
       (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
     if (maybeShuffleMapOutputLoss && !isZombie) {
diff --git a/examples/pom.xml b/examples/pom.xml
index afb21350a0373..3c29aac056258 100644
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
diff --git a/graphx/pom.xml b/graphx/pom.xml
index 65c3caadbf665..9c5592a00e280 100644
--- a/graphx/pom.xml
+++ b/graphx/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
diff --git a/hadoop-cloud/pom.xml b/hadoop-cloud/pom.xml
index 2590417dee78b..bd046552279a6 100644
--- a/hadoop-cloud/pom.xml
+++ b/hadoop-cloud/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
@@ -62,22 +62,23 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-api</artifactId>
       <version>${hadoop.version}</version>
       <scope>provided</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
       <version>${hadoop.version}</version>
+      <scope>${hadoop.deps.scope}</scope>
     </dependency>
     <!--
       the AWS module pulls in jackson; its transitive dependencies can create
       intra-jackson-module version problems.
       -->
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-aws</artifactId>
       <version>${hadoop.version}</version>
       <scope>${hadoop.deps.scope}</scope>
@@ -161,13 +162,13 @@
       <scope>${hadoop.deps.scope}</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-azure</artifactId>
       <version>${hadoop.version}</version>
       <scope>${hadoop.deps.scope}</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-huaweicloud</artifactId>
       <version>${hadoop.version}</version>
       <scope>${huaweicloud.deps.scope}</scope>
@@ -177,7 +178,7 @@
     but it still needs some selective exclusion across versions, especially 3.0.x.
     -->
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-cloud-storage</artifactId>
       <version>${hadoop.version}</version>
       <scope>${hadoop.deps.scope}</scope>
diff --git a/launcher/pom.xml b/launcher/pom.xml
index 60bf4ec1ea7f0..1c614a4ba76ec 100644
--- a/launcher/pom.xml
+++ b/launcher/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
@@ -101,13 +101,13 @@
 
     <!-- Not needed by the test code, but referenced by SparkSubmit which is used by the tests. -->
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-api</artifactId>
       <version>${hadoop.version}</version>
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
       <version>${hadoop.version}</version>
       <scope>test</scope>
diff --git a/mllib-local/pom.xml b/mllib-local/pom.xml
index e1a6db9fdd524..5f56fa42f2478 100644
--- a/mllib-local/pom.xml
+++ b/mllib-local/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
diff --git a/mllib/pom.xml b/mllib/pom.xml
index ec2fe5a599bcf..7469c5f0b1f80 100644
--- a/mllib/pom.xml
+++ b/mllib/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
diff --git a/pom.xml b/pom.xml
index ddccbfb481edc..6cfdb4ce3c432 100644
--- a/pom.xml
+++ b/pom.xml
@@ -26,7 +26,7 @@
   </parent>
   <groupId>org.apache.spark</groupId>
   <artifactId>spark-parent_2.13</artifactId>
-  <version>4.1.3-SNAPSHOT</version>
+  <version>4.1.3.0</version>
   <packaging>pom</packaging>
   <name>Spark Project Parent POM</name>
   <url>https://spark.apache.org/</url>
@@ -127,16 +127,17 @@
     <slf4j.version>2.0.17</slf4j.version>
     <log4j.version>2.24.3</log4j.version>
     <!-- make sure to update IsolatedClientLoader whenever this version is changed -->
-    <hadoop.version>3.4.2</hadoop.version>
+    <hadoop.version>3.4.3.2-EE-SNAPSHOT</hadoop.version>
+    <hadoop.group>io.hops</hadoop.group>
     <!-- SPARK-41247: When updating `protobuf.version`, also need to update `protoVersion` in `SparkBuild.scala` -->
     <protobuf.version>4.33.0</protobuf.version>
     <protoc-jar-maven-plugin.version>3.11.4</protoc-jar-maven-plugin.version>
     <zookeeper.version>3.9.4</zookeeper.version>
     <curator.version>5.9.0</curator.version>
-    <hive.group>org.apache.hive</hive.group>
+    <hive.group>io.hops.hive</hive.group>
     <hive.classifier>core</hive.classifier>
     <!-- Version used in Maven Hive dependency -->
-    <hive.version>2.3.10</hive.version>
+    <hive.version>4.1.0</hive.version>
     <!-- note that this should be compatible with Kafka brokers version 0.10 and up -->
     <kafka.version>3.9.1</kafka.version>
     <!-- After 10.17.1.0, the minimum required version is JDK19 -->
@@ -281,12 +282,14 @@
     <hadoop.deps.scope>compile</hadoop.deps.scope>
     <huaweicloud.deps.scope>compile</huaweicloud.deps.scope>
     <hive.deps.scope>compile</hive.deps.scope>
-    <hive.storage.version>2.8.1</hive.storage.version>
+    <hive.storage.version>2.6.1.2</hive.storage.version>
     <hive.storage.scope>compile</hive.storage.scope>
     <hive.common.scope>compile</hive.common.scope>
     <hive.llap.scope>test</hive.llap.scope>
     <hive.serde.scope>compile</hive.serde.scope>
     <hive.shims.scope>compile</hive.shims.scope>
+    <hive.service.scope>compile</hive.service.scope>
+    <hive.jdbc.scope>compile</hive.jdbc.scope>
     <orc.deps.scope>compile</orc.deps.scope>
     <parquet.deps.scope>compile</parquet.deps.scope>
     <parquet.test.deps.scope>test</parquet.test.deps.scope>
@@ -381,6 +384,28 @@
         <enabled>false</enabled>
       </snapshots>
     </repository>
+    <repository>
+      <id>HopsEE</id>
+      <name>Hops Release Repository</name>
+      <url>https://nexus.hops.works/repository/hops-artifacts</url>
+      <releases>
+        <enabled>true</enabled>
+      </releases>
+      <snapshots>
+        <enabled>true</enabled>
+      </snapshots>
+    </repository>
+    <repository>
+      <id>Hops</id>
+      <name>Hops Repo</name>
+      <url>https://archiva.hops.works/repository/Hops/</url>
+      <releases>
+        <enabled>true</enabled>
+      </releases>
+      <snapshots>
+        <enabled>true</enabled>
+      </snapshots>
+    </repository>
   </repositories>
   <pluginRepositories>
     <pluginRepository>
@@ -460,6 +485,21 @@
       <artifactId>jupiter-interface</artifactId>
       <scope>test</scope>
     </dependency>
+    <!-- Hadoop 3.4.x pulls in reload4j transitively which collides with
+         the log4j-1.2-api bridge and triggers a JVM VerifyError.
+         Mark provided so neither jar is bundled into Spark's assembly. -->
+    <dependency>
+      <groupId>ch.qos.reload4j</groupId>
+      <artifactId>reload4j</artifactId>
+      <version>1.2.22</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-reload4j</artifactId>
+      <version>1.7.36</version>
+      <scope>provided</scope>
+    </dependency>
   </dependencies>
   <dependencyManagement>
     <dependencies>
@@ -1073,6 +1113,11 @@
         <artifactId>xmlschema-core</artifactId>
         <version>${ws.xmlschema.version}</version>
       </dependency>
+      <dependency>
+        <groupId>com.fasterxml.woodstox</groupId>
+        <artifactId>woodstox-core</artifactId>
+        <version>5.4.0</version>
+      </dependency>
       <dependency>
         <groupId>org.glassfish.jersey.core</groupId>
         <artifactId>jersey-server</artifactId>
@@ -1422,13 +1467,13 @@
       </dependency>
       <!-- Hadoop 3.x dependencies -->
       <dependency>
-        <groupId>org.apache.hadoop</groupId>
+        <groupId>${hadoop.group}</groupId>
         <artifactId>hadoop-client-api</artifactId>
         <version>${hadoop.version}</version>
         <scope>${hadoop.deps.scope}</scope>
       </dependency>
       <dependency>
-        <groupId>org.apache.hadoop</groupId>
+        <groupId>${hadoop.group}</groupId>
         <artifactId>hadoop-client-runtime</artifactId>
         <version>${hadoop.version}</version>
         <scope>${hadoop.deps.scope}</scope>
@@ -1437,17 +1482,27 @@
             <groupId>commons-logging</groupId>
             <artifactId>commons-logging</artifactId>
           </exclusion>
+          <!-- reload4j collides with the log4j-1.2-api bridge and triggers
+               a JVM VerifyError; keep it out of Spark's assembly. -->
+          <exclusion>
+            <groupId>ch.qos.reload4j</groupId>
+            <artifactId>reload4j</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-reload4j</artifactId>
+          </exclusion>
         </exclusions>
       </dependency>
       <dependency>
-        <groupId>org.apache.hadoop</groupId>
+        <groupId>${hadoop.group}</groupId>
         <artifactId>hadoop-client-minicluster</artifactId>
         <version>${hadoop.version}</version>
         <scope>test</scope>
       </dependency>
       <!-- End of Hadoop 3.x dependencies -->
       <dependency>
-        <groupId>org.apache.hadoop</groupId>
+        <groupId>${hadoop.group}</groupId>
         <artifactId>hadoop-minikdc</artifactId>
         <version>${hadoop.version}</version>
         <scope>test</scope>
@@ -1722,6 +1777,12 @@
           </exclusion>
         </exclusions>
       </dependency>
+      <dependency>
+        <groupId>${hive.group}</groupId>
+        <artifactId>hive-udf</artifactId>
+        <version>${hive.version}</version>
+        <scope>${hive.deps.scope}</scope>
+      </dependency>
       <dependency>
         <groupId>${hive.group}</groupId>
         <artifactId>hive-common</artifactId>
@@ -1737,11 +1798,11 @@
             <artifactId>ant</artifactId>
           </exclusion>
           <exclusion>
-            <groupId>org.apache.hadoop</groupId>
+            <groupId>${hadoop.group}</groupId>
             <artifactId>hadoop-common</artifactId>
           </exclusion>
           <exclusion>
-            <groupId>org.apache.hadoop</groupId>
+            <groupId>${hadoop.group}</groupId>
             <artifactId>hadoop-auth</artifactId>
           </exclusion>
           <exclusion>
@@ -1944,6 +2005,16 @@
             <groupId>net.hydromatic</groupId>
             <artifactId>aggdesigner-algorithm</artifactId>
           </exclusion>
+          <!-- com.vlkan:flatbuffers is not in Maven Central; use hive-storage-api instead -->
+          <exclusion>
+            <groupId>com.vlkan</groupId>
+            <artifactId>flatbuffers</artifactId>
+          </exclusion>
+          <!-- atlas-intg (pulled in by Hopsworks hive-exec) transitively brings hadoop-common which is banned -->
+          <exclusion>
+            <groupId>org.apache.atlas</groupId>
+            <artifactId>atlas-intg</artifactId>
+          </exclusion>
         </exclusions>
       </dependency>
       <dependency>
@@ -2072,6 +2143,48 @@
         </exclusions>
       </dependency>
 
+      <!--
+        Hive 4.x split RetryUtilities and other metastore server-side classes out of
+        hive-standalone-metastore-common into a new hive-standalone-metastore-server
+        module (didn't exist as a separate artifact in Hive 3.x). Hudi's HiveSyncTool
+        reflects over IMetaStoreClient and eagerly resolves these classes, so this is
+        needed even though Spark only acts as a metastore client.
+      -->
+      <dependency>
+        <groupId>${hive.group}</groupId>
+        <artifactId>hive-standalone-metastore-server</artifactId>
+        <version>${hive.version}</version>
+        <scope>${hive.deps.scope}</scope>
+        <exclusions>
+          <exclusion>
+            <groupId>${hive.group}</groupId>
+            <artifactId>hive-standalone-metastore-common</artifactId>
+          </exclusion>
+          <!-- Already provided (with hadoop-client exclusions) via sql/core's own orc-core dependency -->
+          <exclusion>
+            <groupId>org.apache.orc</groupId>
+            <artifactId>orc-core</artifactId>
+          </exclusion>
+          <!-- Pinned to Hive's old javax.servlet-based Jetty 9; conflicts with Spark's jakarta.servlet-based Jetty 11 -->
+          <exclusion>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-server</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-servlet</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-util</artifactId>
+          </exclusion>
+          <exclusion>
+            <groupId>commons-logging</groupId>
+            <artifactId>commons-logging</artifactId>
+          </exclusion>
+        </exclusions>
+      </dependency>
+
       <dependency>
         <groupId>${hive.group}</groupId>
         <artifactId>hive-serde</artifactId>
@@ -2188,7 +2301,7 @@
             <artifactId>guava</artifactId>
           </exclusion>
           <exclusion>
-            <groupId>org.apache.hadoop</groupId>
+            <groupId>${hadoop.group}</groupId>
             <artifactId>hadoop-yarn-server-resourcemanager</artifactId>
           </exclusion>
           <exclusion>
@@ -3666,4 +3779,10 @@
       </build>
     </profile>
   </profiles>
+  <distributionManagement>
+    <repository>
+      <id>Spark</id>
+      <url>https://nexus.hops.works/repository/hops-artifacts</url>
+    </repository>
+  </distributionManagement>
 </project>
diff --git a/repl/pom.xml b/repl/pom.xml
index 4850b356ec520..703fc038b6b99 100644
--- a/repl/pom.xml
+++ b/repl/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
diff --git a/resource-managers/kubernetes/core/pom.xml b/resource-managers/kubernetes/core/pom.xml
index e2c9966693bb9..87289320ea4cf 100644
--- a/resource-managers/kubernetes/core/pom.xml
+++ b/resource-managers/kubernetes/core/pom.xml
@@ -20,7 +20,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../../pom.xml</relativePath>
   </parent>
 
diff --git a/resource-managers/kubernetes/integration-tests/pom.xml b/resource-managers/kubernetes/integration-tests/pom.xml
index 657ea1bab6f09..cacbce602c9c5 100644
--- a/resource-managers/kubernetes/integration-tests/pom.xml
+++ b/resource-managers/kubernetes/integration-tests/pom.xml
@@ -20,7 +20,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../../pom.xml</relativePath>
   </parent>
 
diff --git a/resource-managers/yarn/pom.xml b/resource-managers/yarn/pom.xml
index 90601a98fd5d4..fef5c38730b21 100644
--- a/resource-managers/yarn/pom.xml
+++ b/resource-managers/yarn/pom.xml
@@ -20,7 +20,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -64,19 +64,19 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-api</artifactId>
       <version>${hadoop.version}</version>
       <scope>${hadoop.deps.scope}</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
       <version>${hadoop.version}</version>
       <scope>${hadoop.deps.scope}</scope>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-minicluster</artifactId>
       <version>${hadoop.version}</version>
       <scope>test</scope>
diff --git a/sql/api/pom.xml b/sql/api/pom.xml
index 185ca1d1f317c..612f61f1c0d57 100644
--- a/sql/api/pom.xml
+++ b/sql/api/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-parent_2.13</artifactId>
-        <version>4.1.3-SNAPSHOT</version>
+        <version>4.1.3.0</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
 
diff --git a/sql/catalyst/pom.xml b/sql/catalyst/pom.xml
index 3916628203e98..b13d15244e546 100644
--- a/sql/catalyst/pom.xml
+++ b/sql/catalyst/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -134,6 +134,10 @@
       <groupId>org.apache.ws.xmlschema</groupId>
       <artifactId>xmlschema-core</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.woodstox</groupId>
+      <artifactId>woodstox-core</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.datasketches</groupId>
       <artifactId>datasketches-java</artifactId>
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala
index 8793eebc97202..f77349f4c1396 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXMLRecordReader.scala
@@ -23,7 +23,7 @@ import javax.xml.transform.stax.StAXSource
 
 import scala.util.control.NonFatal
 
-import org.apache.hadoop.shaded.com.ctc.wstx.exc.WstxEOFException
+import com.ctc.wstx.exc.WstxEOFException
 
 import org.apache.spark.internal.Logging
 import org.apache.spark.util.SparkErrorUtils
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlGenerator.scala
index b6a4d74a8e979..74b543990b222 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlGenerator.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/StaxXmlGenerator.scala
@@ -23,7 +23,7 @@ import javax.xml.stream.XMLOutputFactory
 
 import scala.collection.Map
 
-import org.apache.hadoop.shaded.com.ctc.wstx.api.WstxOutputProperties
+import com.ctc.wstx.api.WstxOutputProperties
 
 import org.apache.spark.SparkIllegalArgumentException
 import org.apache.spark.sql.catalyst.InternalRow
diff --git a/sql/connect/client/jdbc/pom.xml b/sql/connect/client/jdbc/pom.xml
index 54a87d99998d4..9da6e8d33ee97 100644
--- a/sql/connect/client/jdbc/pom.xml
+++ b/sql/connect/client/jdbc/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../../../pom.xml</relativePath>
   </parent>
 
diff --git a/sql/connect/client/jvm/pom.xml b/sql/connect/client/jvm/pom.xml
index dbad052e11b17..de42d4ca61d9f 100644
--- a/sql/connect/client/jvm/pom.xml
+++ b/sql/connect/client/jvm/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../../../pom.xml</relativePath>
   </parent>
 
diff --git a/sql/connect/common/pom.xml b/sql/connect/common/pom.xml
index 2e8b2c0c57ce2..99240289d76b9 100644
--- a/sql/connect/common/pom.xml
+++ b/sql/connect/common/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-parent_2.13</artifactId>
-        <version>4.1.3-SNAPSHOT</version>
+        <version>4.1.3.0</version>
         <relativePath>../../../pom.xml</relativePath>
     </parent>
 
diff --git a/sql/connect/server/pom.xml b/sql/connect/server/pom.xml
index ac9a3a4c9b103..12cf672e14da0 100644
--- a/sql/connect/server/pom.xml
+++ b/sql/connect/server/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../../pom.xml</relativePath>
   </parent>
 
diff --git a/sql/connect/shims/pom.xml b/sql/connect/shims/pom.xml
index d114b0a931c12..93cc52e55d577 100644
--- a/sql/connect/shims/pom.xml
+++ b/sql/connect/shims/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../../pom.xml</relativePath>
   </parent>
 
diff --git a/sql/core/pom.xml b/sql/core/pom.xml
index 8d211d7afdb6f..94ca5844e435c 100644
--- a/sql/core/pom.xml
+++ b/sql/core/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -122,16 +122,64 @@
       <groupId>org.apache.orc</groupId>
       <artifactId>orc-core</artifactId>
       <classifier>${orc.classifier}</classifier>
+      <exclusions>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-client-runtime</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-client-api</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>org.apache.orc</groupId>
       <artifactId>orc-mapreduce</artifactId>
       <classifier>${orc.classifier}</classifier>
+      <exclusions>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-client-api</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-client-runtime</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>${hive.group}</groupId>
       <artifactId>hive-storage-api</artifactId>
     </dependency>
+    <dependency>
+      <groupId>${hive.group}</groupId>
+      <artifactId>hive-common</artifactId>
+      <scope>${hive.common.scope}</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-auth</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-annotations</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-yarn-api</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-yarn-client</artifactId>
+        </exclusion>
+      </exclusions>
+    </dependency>
+    <dependency>
+      <groupId>${hive.group}</groupId>
+      <artifactId>hive-serde</artifactId>
+      <scope>${hive.serde.scope}</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.parquet</groupId>
       <artifactId>parquet-column</artifactId>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ShuffledRowRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ShuffledRowRDD.scala
index 367732dbb2059..2276dbb7b8f92 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/ShuffledRowRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/ShuffledRowRDD.scala
@@ -166,6 +166,9 @@ class ShuffledRowRDD(
   }
 
   override def getPreferredLocations(partition: Partition): Seq[String] = {
+    if (sparkContext.conf.isRssEnable()) {
+      return Nil
+    }
     val tracker = SparkEnv.get.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
     partition.asInstanceOf[ShuffledRowRDDPartition].spec match {
       case CoalescedPartitionSpec(startReducerIndex, endReducerIndex, _) =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DaysWritableV2.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DaysWritableV2.scala
new file mode 100644
index 0000000000000..3133d7b12eb2f
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DaysWritableV2.scala
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.spark.sql.execution.datasources
+
+import java.io.{DataInput, DataOutput, IOException}
+
+import org.apache.hadoop.hive.common.`type`.Date
+import org.apache.hadoop.hive.serde2.io.DateWritableV2
+import org.apache.hadoop.io.WritableUtils
+
+import org.apache.spark.sql.catalyst.util.RebaseDateTime.{rebaseGregorianToJulianDays, rebaseJulianToGregorianDays}
+
+
+/**
+ * The class accepts/returns days in Gregorian calendar and rebase them
+ * via conversion to local date in Julian calendar for dates before 1582-10-15
+ * in read/write for backward compatibility with Spark 2.4 and earlier versions.
+ *
+ * @param gregorianDays The number of days since the epoch 1970-01-01 in
+ *                      Gregorian calendar.
+ * @param julianDays The number of days since the epoch 1970-01-01 in
+ *                   Julian calendar.
+ */
+class DaysWritableV2(
+    var gregorianDays: Int,
+    var julianDays: Int)
+  extends DateWritableV2 {
+
+  def this() = this(0, 0)
+  def this(gregorianDays: Int) =
+    this(gregorianDays, rebaseGregorianToJulianDays(gregorianDays))
+  def this(dateWritable: DateWritableV2) = {
+    this(
+      gregorianDays = dateWritable match {
+        case daysWritable: DaysWritableV2 => daysWritable.gregorianDays
+        case dateWritable: DateWritableV2 =>
+        rebaseJulianToGregorianDays(dateWritable.getDays)
+      },
+      julianDays = dateWritable.getDays)
+  }
+
+  override def getDays: Int = julianDays
+  override def get: Date = {
+    Date.ofEpochMilli(DateWritableV2.daysToMillis(julianDays))
+  }
+
+  override def set(d: Int): Unit = {
+    gregorianDays = d
+    julianDays = rebaseGregorianToJulianDays(d)
+  }
+
+  @throws[IOException]
+  override def write(out: DataOutput): Unit = {
+    WritableUtils.writeVInt(out, julianDays)
+  }
+
+  @throws[IOException]
+  override def readFields(in: DataInput): Unit = {
+    julianDays = WritableUtils.readVInt(in)
+    gregorianDays = rebaseJulianToGregorianDays(julianDays)
+  }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XSDToSchema.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XSDToSchema.scala
index c03c0ba11de57..04afc30aa6577 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XSDToSchema.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/xml/XSDToSchema.scala
@@ -16,12 +16,11 @@
  */
 package org.apache.spark.sql.execution.datasources.xml
 
-import java.io.StringReader
+import java.io.{InputStreamReader, StringReader}
 
 import scala.jdk.CollectionConverters._
 
 import org.apache.hadoop.fs.Path
-import org.apache.hadoop.shaded.org.jline.utils.InputStreamReader
 import org.apache.ws.commons.schema._
 import org.apache.ws.commons.schema.constants.Constants
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala
index bf9740970a667..6947a2e4f68e9 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcFilterSuite.scala
@@ -90,7 +90,7 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
       (implicit df: DataFrame): Unit = {
     def checkLogicalOperator(filter: SearchArgument) = {
       // HIVE-24458 changes toString output and provides `toOldString` for old style.
-      assert(filter.asInstanceOf[SearchArgumentImpl].toOldString == stringExpr)
+      assert(filter.asInstanceOf[SearchArgumentImpl].toString == stringExpr)
     }
     checkFilterPredicate(df, predicate, checkLogicalOperator)
   }
@@ -547,7 +547,7 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
       OrcFilters.createFilter(schema, Array(
         LessThan("a", 10),
         StringContains("b", "prefix")
-      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toOldString
+      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toString
     }
 
     // The `LessThan` should be converted while the whole inner `And` shouldn't
@@ -558,7 +558,7 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
           GreaterThan("a", 1),
           StringContains("b", "prefix")
         ))
-      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toOldString
+      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toString
     }
 
     // Safely remove unsupported `StringContains` predicate and push down `LessThan`
@@ -568,7 +568,7 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
           LessThan("a", 10),
           StringContains("b", "prefix")
         )
-      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toOldString
+      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toString
     }
 
     // Safely remove unsupported `StringContains` predicate, push down `LessThan` and `GreaterThan`.
@@ -582,7 +582,7 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
           ),
           GreaterThan("a", 1)
         )
-      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toOldString
+      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toString
     }
   }
 
@@ -605,7 +605,7 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
             LessThan("a", 1)
           )
         )
-      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toOldString
+      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toString
     }
 
     assertResult("leaf-0 = (LESS_THAN_EQUALS a 10), leaf-1 = (LESS_THAN a 1)," +
@@ -621,7 +621,7 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
             LessThan("a", 1)
           )
         )
-      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toOldString
+      ).toImmutableArraySeq).get.asInstanceOf[SearchArgumentImpl].toString
     }
 
     assert(OrcFilters.createFilter(schema, Array(
@@ -643,7 +643,7 @@ class OrcFilterSuite extends OrcTest with SharedSparkSession {
         LessThan(
           "a",
           new java.math.BigDecimal(3.14, MathContext.DECIMAL64).setScale(2))).toImmutableArraySeq
-      ).get.asInstanceOf[SearchArgumentImpl].toOldString
+      ).get.asInstanceOf[SearchArgumentImpl].toString
     }
   }
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcV1FilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcV1FilterSuite.scala
index 8018417f923af..11e1e9d3ad263 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcV1FilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/orc/OrcV1FilterSuite.scala
@@ -80,7 +80,7 @@ class OrcV1FilterSuite extends OrcFilterSuite {
       (implicit df: DataFrame): Unit = {
     def checkLogicalOperator(filter: SearchArgument) = {
       // HIVE-24458 changes toString format and provides `toOldString` for old style.
-      assert(filter.asInstanceOf[SearchArgumentImpl].toOldString == stringExpr)
+      assert(filter.asInstanceOf[SearchArgumentImpl].toString == stringExpr)
     }
     checkFilterPredicate(df, predicate, checkLogicalOperator)
   }
diff --git a/sql/hive-thriftserver/pom.xml b/sql/hive-thriftserver/pom.xml
index f088304054d14..115184bf79ca6 100644
--- a/sql/hive-thriftserver/pom.xml
+++ b/sql/hive-thriftserver/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
diff --git a/sql/hive/pom.xml b/sql/hive/pom.xml
index d5f1129ec7bf3..c40f0a67e64a8 100644
--- a/sql/hive/pom.xml
+++ b/sql/hive/pom.xml
@@ -22,7 +22,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 
@@ -108,16 +108,47 @@
       <groupId>${hive.group}</groupId>
       <artifactId>hive-common</artifactId>
       <scope>${hive.common.scope}</scope>
+      <exclusions>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-auth</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-annotations</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-yarn-api</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.apache.hadoop</groupId>
+          <artifactId>hadoop-yarn-client</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>${hive.group}</groupId>
       <artifactId>hive-exec</artifactId>
       <classifier>${hive.classifier}</classifier>
     </dependency>
+    <dependency>
+      <groupId>${hive.group}</groupId>
+      <artifactId>hive-udf</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>${hive.group}</groupId>
+      <artifactId>hive-storage-api</artifactId>
+      <version>${hive.version}</version>
+    </dependency>
     <dependency>
       <groupId>${hive.group}</groupId>
       <artifactId>hive-metastore</artifactId>
     </dependency>
+    <dependency>
+      <groupId>${hive.group}</groupId>
+      <artifactId>hive-standalone-metastore-server</artifactId>
+    </dependency>
     <dependency>
       <groupId>${hive.group}</groupId>
       <artifactId>hive-serde</artifactId>
@@ -173,7 +204,7 @@
       <artifactId>datanucleus-core</artifactId>
     </dependency>
     <dependency>
-      <groupId>org.apache.hadoop</groupId>
+      <groupId>${hadoop.group}</groupId>
       <artifactId>hadoop-client-runtime</artifactId>
       <scope>${hadoop.deps.scope}</scope>
     </dependency>
@@ -269,6 +300,50 @@
           </execution>
         </executions>
       </plugin>
+      <!--
+        hive-exec's `core` classifier (see hive.classifier) strips out Hive's own ANTLR-generated
+        SQL parser and a couple of small `common`-package classes, since Spark uses its own SQL
+        parser and doesn't need them. However, Hudi's HiveSyncTool (via IMetaStoreClientUtil.getMSC)
+        reflects over org.apache.hadoop.hive.ql.metadata.Hive with Class#getMethod, which eagerly
+        resolves every type referenced across ALL of that class's method signatures - including
+        ASTNode and friends - even though none of those methods are actually invoked. Missing any
+        of them throws NoClassDefFoundError instead of the NoSuchMethodException the reflection
+        code's catch block is written to expect. Pull just this handful of classes out of the full
+        (non-core) hive-exec jar rather than depending on the whole 90MB+ shaded artifact, which
+        would reintroduce the bundled hadoop/orc/calcite/tez duplicate-class problems worked around
+        elsewhere in this pom.
+      -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-dependency-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>unpack-hive-exec-ql-shims</id>
+            <phase>generate-resources</phase>
+            <goals>
+              <goal>unpack</goal>
+            </goals>
+            <configuration>
+              <artifactItems>
+                <artifactItem>
+                  <groupId>${hive.group}</groupId>
+                  <artifactId>hive-exec</artifactId>
+                  <version>${hive.version}</version>
+                  <outputDirectory>${project.build.outputDirectory}</outputDirectory>
+                  <includes>
+                    org/apache/hadoop/hive/ql/parse/ASTNode.class,
+                    org/apache/hadoop/hive/ql/parse/ASTNodeOrigin.class,
+                    org/apache/hadoop/hive/common/MaterializationSnapshot.class,
+                    org/apache/hadoop/hive/common/TableName.class,
+                    org/apache/hadoop/hive/common/type/SnapshotContext.class,
+                    org/apache/hadoop/hive/common/type/SnapshotContext$WriteOperationType.class
+                  </includes>
+                </artifactItem>
+              </artifactItems>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
 </project>
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
index 087f68ce3c790..b4c06b3807646 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveInspectors.scala
@@ -23,7 +23,7 @@ import java.time.Duration
 import scala.jdk.CollectionConverters._
 
 import org.apache.hadoop.{io => hadoopIo}
-import org.apache.hadoop.hive.common.`type`.{HiveChar, HiveDecimal, HiveIntervalDayTime, HiveIntervalYearMonth, HiveVarchar}
+import org.apache.hadoop.hive.common.`type`.{HiveChar, HiveDecimal, HiveIntervalDayTime, HiveIntervalYearMonth, HiveVarchar, Timestamp}
 import org.apache.hadoop.hive.serde2.{io => hiveIo}
 import org.apache.hadoop.hive.serde2.objectinspector.{StructField => HiveStructField, _}
 import org.apache.hadoop.hive.serde2.objectinspector.primitive._
@@ -35,7 +35,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.errors.DataTypeErrors.toSQLType
-import org.apache.spark.sql.execution.datasources.DaysWritable
+import org.apache.spark.sql.execution.datasources.DaysWritableV2
 import org.apache.spark.sql.types
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
@@ -191,8 +191,8 @@ private[hive] trait HiveInspectors {
     case c: Class[_] if c == classOf[hiveIo.HiveDecimalWritable] => DecimalType.SYSTEM_DEFAULT
     case c: Class[_] if c == classOf[hiveIo.ByteWritable] => ByteType
     case c: Class[_] if c == classOf[hiveIo.ShortWritable] => ShortType
-    case c: Class[_] if c == classOf[hiveIo.DateWritable] => DateType
-    case c: Class[_] if c == classOf[hiveIo.TimestampWritable] => TimestampType
+    case c: Class[_] if c == classOf[hiveIo.DateWritableV2] => DateType
+    case c: Class[_] if c == classOf[hiveIo.TimestampWritableV2] => TimestampType
     case c: Class[_] if c == classOf[hadoopIo.Text] => StringType
     case c: Class[_] if c == classOf[hadoopIo.IntWritable] => IntegerType
     case c: Class[_] if c == classOf[hadoopIo.LongWritable] => LongType
@@ -480,7 +480,8 @@ private[hive] trait HiveInspectors {
         _ => constant
       case poi: WritableConstantTimestampObjectInspector =>
         val t = poi.getWritableConstantValue
-        val constant = DateTimeUtils.fromJavaTimestamp(t.getTimestamp)
+        val constant = DateTimeUtils.fromJavaTimestamp(
+          new java.sql.Timestamp(t.getTimestamp.toEpochMilli))
         _ => constant
       case poi: WritableConstantIntObjectInspector =>
         val constant = poi.getWritableConstantValue.get()
@@ -509,7 +510,8 @@ private[hive] trait HiveInspectors {
         System.arraycopy(writable.getBytes, 0, constant, 0, constant.length)
         _ => constant
       case poi: WritableConstantDateObjectInspector =>
-        val constant = DateTimeUtils.fromJavaDate(poi.getWritableConstantValue.get())
+        val constant = DateTimeUtils.fromJavaDate(new java.sql.Date(poi.getWritableConstantValue
+          .get().toEpochMilli))
         _ => constant
       case mi: StandardConstantMapObjectInspector =>
         val keyUnwrapper = unwrapperFor(mi.getMapKeyObjectInspector)
@@ -633,7 +635,7 @@ private[hive] trait HiveInspectors {
         case x: DateObjectInspector if x.preferWritable() =>
           data: Any => {
             if (data != null) {
-              new DaysWritable(x.getPrimitiveWritableObject(data)).gregorianDays
+              new DaysWritableV2(x.getPrimitiveWritableObject(data).getDays).gregorianDays
             } else {
               null
             }
@@ -641,7 +643,8 @@ private[hive] trait HiveInspectors {
         case x: DateObjectInspector =>
           data: Any => {
             if (data != null) {
-              DateTimeUtils.fromJavaDate(x.getPrimitiveJavaObject(data))
+              DateTimeUtils.fromJavaDate(new java.sql.Date(x.getPrimitiveJavaObject(data)
+                .toEpochMilli))
             } else {
               null
             }
@@ -649,7 +652,8 @@ private[hive] trait HiveInspectors {
         case x: TimestampObjectInspector if x.preferWritable() =>
           data: Any => {
             if (data != null) {
-              DateTimeUtils.fromJavaTimestamp(x.getPrimitiveWritableObject(data).getTimestamp)
+              DateTimeUtils.fromJavaTimestamp(new java.sql.Timestamp(
+                x.getPrimitiveWritableObject(data).getTimestamp.toEpochMilli))
             } else {
               null
             }
@@ -657,7 +661,8 @@ private[hive] trait HiveInspectors {
         case ti: TimestampObjectInspector =>
           data: Any => {
             if (data != null) {
-              DateTimeUtils.fromJavaTimestamp(ti.getPrimitiveJavaObject(data))
+              DateTimeUtils.fromJavaTimestamp(
+                new java.sql.Timestamp(ti.getPrimitiveJavaObject(data).toEpochMilli))
             } else {
               null
             }
@@ -1094,18 +1099,19 @@ private[hive] trait HiveInspectors {
       new hadoopIo.BytesWritable(value.asInstanceOf[Array[Byte]])
     }
 
-  private def getDateWritable(value: Any): DaysWritable =
+  private def getDateWritable(value: Any): DaysWritableV2 =
     if (value == null) {
       null
     } else {
-      new DaysWritable(value.asInstanceOf[Int])
+      new DaysWritableV2(value.asInstanceOf[Int])
     }
 
-  private def getTimestampWritable(value: Any): hiveIo.TimestampWritable =
+  private def getTimestampWritable(value: Any): hiveIo.TimestampWritableV2 =
     if (value == null) {
       null
     } else {
-      new hiveIo.TimestampWritable(DateTimeUtils.toJavaTimestamp(value.asInstanceOf[Long]))
+      val ts = DateTimeUtils.toJavaTimestamp(value.asInstanceOf[Long])
+      new hiveIo.TimestampWritableV2(Timestamp.ofEpochMilli(ts.getTime))
     }
 
   private def getHiveIntervalDayTimeWritable(value: Any): hiveIo.HiveIntervalDayTimeWritable =
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
index d6f62284d2e7c..d6c4d6045382f 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala
@@ -27,7 +27,7 @@ import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants._
 import org.apache.hadoop.hive.ql.exec.Utilities
 import org.apache.hadoop.hive.ql.metadata.{Partition => HivePartition, Table => HiveTable}
 import org.apache.hadoop.hive.ql.plan.{PartitionDesc, TableDesc}
-import org.apache.hadoop.hive.serde2.Deserializer
+import org.apache.hadoop.hive.serde2.{AbstractSerDe, Deserializer}
 import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils.AvroTableProperties
 import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspectorConverters, StructObjectInspector}
 import org.apache.hadoop.hive.serde2.objectinspector.primitive._
@@ -135,7 +135,8 @@ class HadoopTableReader(
       val hconf = broadcastedHadoopConf.value.value
       val deserializer = deserializerClass.getConstructor().newInstance()
       DeserializerLock.synchronized {
-        deserializer.initialize(hconf, localTableDesc.getProperties)
+        deserializer.asInstanceOf[AbstractSerDe]
+          .initialize(hconf, localTableDesc.getProperties, null)
       }
       HadoopTableReader.fillObject(iter, deserializer, attrsWithIndex, mutableRow, deserializer)
     }
@@ -226,13 +227,10 @@ class HadoopTableReader(
           case (key, value) => props.setProperty(key, value)
         }
         DeserializerLock.synchronized {
-          deserializer.initialize(hconf, props)
+          deserializer.asInstanceOf[AbstractSerDe].initialize(hconf, props, null)
         }
         // get the table deserializer
-        val tableSerDe = localTableDesc.getDeserializerClass.getConstructor().newInstance()
-        DeserializerLock.synchronized {
-          tableSerDe.initialize(hconf, tableProperties)
-        }
+        val tableSerDe = localTableDesc.getDeserializer(hconf)
 
         // fill the non partition key attributes
         HadoopTableReader.fillObject(iter, deserializer, nonPartitionKeyAttrs,
@@ -488,10 +486,12 @@ private[hive] object HadoopTableReader extends HiveInspectors with Logging {
             row.update(ordinal, HiveShim.toCatalystDecimal(oi, value))
         case oi: TimestampObjectInspector =>
           (value: Any, row: InternalRow, ordinal: Int) =>
-            row.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(oi.getPrimitiveJavaObject(value)))
+            row.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(new java.sql.Timestamp(
+              oi.getPrimitiveJavaObject(value).toEpochMilli)))
         case oi: DateObjectInspector =>
           (value: Any, row: InternalRow, ordinal: Int) =>
-            row.setInt(ordinal, DateTimeUtils.fromJavaDate(oi.getPrimitiveJavaObject(value)))
+            row.setInt(ordinal, DateTimeUtils.fromJavaDate(new java.sql.Date(oi
+              .getPrimitiveJavaObject(value).toEpochMilli)))
         case oi: BinaryObjectInspector =>
           (value: Any, row: InternalRow, ordinal: Int) =>
             row.update(ordinal, oi.getPrimitiveJavaObject(value))
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index be46e54d18ec1..52a150c27def2 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -37,7 +37,6 @@ import org.apache.hadoop.hive.metastore.{HiveMetaStoreClient, IMetaStoreClient,
 import org.apache.hadoop.hive.metastore.api.{Database => HiveDatabase, Table => MetaStoreApiTable, _}
 import org.apache.hadoop.hive.ql.Driver
 import org.apache.hadoop.hive.ql.metadata.{Hive, HiveException, Partition => HivePartition, Table => HiveTable}
-import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer.HIVE_COLUMN_ORDER_ASC
 import org.apache.hadoop.hive.ql.processors._
 import org.apache.hadoop.hive.ql.session.SessionState
 import org.apache.hadoop.hive.serde.serdeConstants
@@ -473,7 +472,7 @@ private[hive] class HiveClientImpl(
       // are sorted in ascending order, only then propagate the sortedness information
       // to downstream processing / optimizations in Spark
       // TODO: In future we can have Spark support columns sorted in descending order
-      val allAscendingSorted = sortColumnOrders.forall(_.getOrder == HIVE_COLUMN_ORDER_ASC)
+      val allAscendingSorted = sortColumnOrders.forall(_.getOrder == 1)
 
       val sortColumnNames = if (allAscendingSorted) {
         sortColumnOrders.map(_.getCol)
@@ -875,20 +874,11 @@ private[hive] class HiveClientImpl(
       // Since HIVE-18238(Hive 3.0.0), the Driver.close function's return type changed
       // and the CommandProcessorFactory.clean function removed.
       driver.getClass.getMethod("close").invoke(driver)
-      if (version < hive.v3_0) {
-        CommandProcessorFactory.clean(conf)
-      }
     }
 
     def getResponseCode(response: CommandProcessorResponse): Int = {
-      if (version < hive.v4_0) {
-        response.getResponseCode
-      } else {
-        // Since Hive 4.0, response code is removed from CommandProcessorResponse.
-        // Here we simply return 0 for the positive cases as for error cases it will
-        // throw exceptions early.
-        0
-      }
+      // Since Hive 4.0, response code is removed from CommandProcessorResponse.
+      0
     }
 
     // Hive query needs to start SessionState.
@@ -908,7 +898,7 @@ private[hive] class HiveClientImpl(
             if (getResponseCode(response) != 0) {
               // Throw an exception if there is an error in query processing.
               // This works for hive 3.x and earlier versions.
-              throw new QueryExecutionException(response.getErrorMessage)
+              throw new QueryExecutionException(response.getMessage)
             }
             driver.setMaxRows(maxRows)
             val results = shim.getDriverResults(driver)
@@ -938,7 +928,7 @@ private[hive] class HiveClientImpl(
             // Throw an exception if there is an error in query processing.
             // This works for hive 3.x and earlier versions. For 4.x and later versions,
             // It will go to the catch block directly.
-            throw new QueryExecutionException(response.getErrorMessage)
+            throw new QueryExecutionException(response.getMessage)
           }
           Seq(responseCode.toString)
       }
@@ -1062,18 +1052,7 @@ private[hive] class HiveClientImpl(
     others.foreach { table =>
       val t = table.getTableName
       logDebug(s"Deleting table $t")
-      try {
-        shim.getIndexes(client, "default", t, 255).foreach { index =>
-          shim.dropIndex(client, "default", t, index.getIndexName)
-        }
-        if (!table.isIndexTable) {
-          shim.dropTable(client, "default", t)
-        }
-      } catch {
-        case _: NoSuchMethodError =>
-          // HIVE-18448 Hive 3.0 remove index APIs
-          shim.dropTable(client, "default", t)
-      }
+      shim.dropTable(client, "default", t)
     }
     shim.getAllDatabases(client).filterNot(_ == "default").foreach { db =>
       logDebug(s"Dropping Database: $db")
@@ -1205,7 +1184,7 @@ private[hive] object HiveClientImpl extends Logging {
         if (bucketSpec.sortColumnNames.nonEmpty) {
           hiveTable.setSortCols(
             bucketSpec.sortColumnNames
-              .map(col => new Order(col, HIVE_COLUMN_ORDER_ASC))
+              .map(col => new Order(col, 1))
               .toList
               .asJava
           )
@@ -1404,8 +1383,8 @@ private[hive] object HiveClientImpl extends Logging {
 
   /**
    * Initialize Hive through Configuration.
-   * First try to use getWithoutRegisterFns to initialize to avoid loading all functions,
-   * if there is no such method, fallback to Hive.get.
+   * For Hops Hive 3.0, use getWithFastCheck; fall back to getWithoutRegisterFns or Hive.get
+   * for other versions.
    */
   def getHive(conf: Configuration): Hive = {
     val hiveConf = conf match {
@@ -1415,12 +1394,17 @@ private[hive] object HiveClientImpl extends Logging {
         new HiveConf(conf, classOf[HiveConf])
     }
     val hive = try {
-      Hive.getWithoutRegisterFns(hiveConf)
+      Hive.getWithFastCheck(hiveConf, false)
     } catch {
-      // SPARK-37069: not all Hive versions have the above method (e.g., Hive 2.3.9 has it but
-      // 2.3.8 doesn't), therefore here we fallback when encountering the exception.
       case _: NoSuchMethodError =>
-        Hive.get(hiveConf)
+        try {
+          Hive.getWithoutRegisterFns(hiveConf)
+        } catch {
+          // SPARK-37069: not all Hive versions have the above method (e.g., Hive 2.3.9 has it but
+          // 2.3.8 doesn't), therefore here we fallback when encountering the exception.
+          case _: NoSuchMethodError =>
+            Hive.get(hiveConf)
+        }
     }
 
     // Follow behavior of HIVE-26633 (4.0.0), only apply the max message size when
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala
index ef27669f5ba09..13f49653e1ffc 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala
@@ -29,11 +29,12 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.hive.conf.HiveConf
 import org.apache.hadoop.hive.metastore.{IMetaStoreClient, PartitionDropOptions, TableType}
-import org.apache.hadoop.hive.metastore.api.{Database, EnvironmentContext, Function => HiveFunction, FunctionType, Index, MetaException, PrincipalType, ResourceType, ResourceUri}
+import org.apache.hadoop.hive.metastore.api.{Database, EnvironmentContext, Function => HiveFunction, FunctionType, MetaException, PrincipalType, ResourceType, ResourceUri}
 import org.apache.hadoop.hive.ql.Driver
+import org.apache.hadoop.hive.ql.ddl.table.partition.add.AlterTableAddPartitionDesc
 import org.apache.hadoop.hive.ql.io.AcidUtils
 import org.apache.hadoop.hive.ql.metadata.{Hive, HiveException, Partition, Table}
-import org.apache.hadoop.hive.ql.plan.{AddPartitionDesc, DynamicPartitionCtx}
+import org.apache.hadoop.hive.ql.plan.DynamicPartitionCtx
 import org.apache.hadoop.hive.ql.processors.{CommandProcessor, CommandProcessorFactory}
 import org.apache.hadoop.hive.ql.session.SessionState
 import org.apache.hadoop.hive.serde.serdeConstants
@@ -211,7 +212,8 @@ private[client] sealed abstract class Shim {
 
   def listFunctions(hive: Hive, db: String, pattern: String): Seq[String]
 
-  def dropIndex(hive: Hive, dbName: String, tableName: String, indexName: String): Unit
+  // def dropIndex(hive: Hive, dbName: String, tableName: String, indexName: String): Unit
+  // Hops Hive 3.0 removed Index APIs; dropIndex is no longer available.
 
   def dropTable(
       hive: Hive,
@@ -235,7 +237,8 @@ private[client] sealed abstract class Shim {
 
   def getMSC(hive: Hive): IMetaStoreClient
 
-  def getIndexes(hive: Hive, dbName: String, tableName: String, max: Short): Seq[Index]
+  // def getIndexes(hive: Hive, dbName: String, tableName: String, max: Short): Seq[Index]
+  // Hops Hive 3.0 removed Index APIs; getIndexes is no longer available.
 
   protected def findMethod(klass: Class[_], name: String, args: Class[_]*): Method = {
     klass.getMethod(name, args: _*)
@@ -324,16 +327,18 @@ private[client] class Shim_v2_0 extends Shim with Logging {
       table: Table,
       parts: Seq[CatalogTablePartition],
       ignoreIfExists: Boolean): Unit = {
-    val addPartitionDesc = new AddPartitionDesc(table.getDbName, table.getTableName, ignoreIfExists)
-    parts.zipWithIndex.foreach { case (s, i) =>
-      addPartitionDesc.addPartition(
-        s.spec.asJava, s.storage.locationUri.map(CatalogUtils.URIToString).orNull)
-      if (s.parameters.nonEmpty) {
-        addPartitionDesc.getPartition(i).setPartParams(s.parameters.asJava)
-      }
-    }
+    val partDescs = parts.map { s =>
+      new AlterTableAddPartitionDesc.PartitionDesc(
+        s.spec.asJava,
+        s.storage.locationUri.map(CatalogUtils.URIToString).orNull,
+        s.parameters.asJava)
+    }.asJava
+    val addPartitionDesc = new AlterTableAddPartitionDesc(
+      table.getDbName, table.getTableName, ignoreIfExists, partDescs)
     recordHiveCall()
-    hive.createPartitions(addPartitionDesc)
+    addPartitionDesc.getPartitions.asScala.foreach { pd =>
+      hive.createPartition(table, pd.getPartSpec)
+    }
   }
 
   override def getAllPartitions(hive: Hive, table: Table): Seq[Partition] = {
@@ -516,10 +521,12 @@ private[client] class Shim_v2_0 extends Shim with Logging {
       numDP: JInteger, listBucketingEnabled: JBoolean, isAcid, txnIdInLoadDynamicPartitions)
   }
 
+  /* Hops Hive 3.0 removed Index APIs
   override def dropIndex(hive: Hive, dbName: String, tableName: String, indexName: String): Unit = {
     recordHiveCall()
     hive.dropIndex(dbName, tableName, indexName, throwExceptionInDropIndex, deleteDataInDropIndex)
   }
+  */
 
   override def dropTable(
       hive: Hive,
@@ -995,9 +1002,10 @@ private[client] class Shim_v2_0 extends Shim with Logging {
       oldPartSpec: JMap[String, String],
       newPart: Partition): Unit = {
     recordHiveCall()
-    hive.renamePartition(table, oldPartSpec, newPart)
+    hive.renamePartition(table, oldPartSpec, newPart, 0L)
   }
 
+  /* Hops Hive 3.0 removed Index APIs
   override def getIndexes(
       hive: Hive,
       dbName: String,
@@ -1006,6 +1014,7 @@ private[client] class Shim_v2_0 extends Shim with Logging {
     recordHiveCall()
     hive.getIndexes(dbName, tableName, max).asScala.toSeq
   }
+  */
 }
 
 private[client] class Shim_v2_1 extends Shim_v2_0 {
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/IsolatedClientLoader.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/IsolatedClientLoader.scala
index c439dfbd91692..db7e24b1cada7 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/IsolatedClientLoader.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/IsolatedClientLoader.scala
@@ -90,18 +90,20 @@ private[hive] object IsolatedClientLoader extends Logging {
   }
 
   def hiveVersion(version: String): HiveVersion = {
-    VersionUtils.majorMinorPatchVersion(version).flatMap {
-      case (2, 0, _) => Some(hive.v2_0)
-      case (2, 1, _) => Some(hive.v2_1)
-      case (2, 2, _) => Some(hive.v2_2)
-      case (2, 3, _) => Some(hive.v2_3)
-      case (3, 0, _) => Some(hive.v3_0)
-      case (3, 1, _) => Some(hive.v3_1)
-      case (4, 0, _) => Some(hive.v4_0)
-      case (4, 1, _) => Some(hive.v4_1)
-      case _ => None
-    }.getOrElse {
-      throw QueryExecutionErrors.unsupportedHiveMetastoreVersionError(
+    // Hops Hive versions like "3.0.0.14.5" have extra segments that VersionUtils cannot parse.
+    // Extract only major.minor for matching.
+    val parts = version.split("\\.")
+    val majorMinor = if (parts.length >= 2) s"${parts(0)}.${parts(1)}" else parts(0)
+    majorMinor match {
+      case "2.0" => hive.v2_0
+      case "2.1" => hive.v2_1
+      case "2.2" => hive.v2_2
+      case "2.3" => hive.v2_3
+      case "3.0" => hive.v3_0
+      case "3.1" => hive.v3_1
+      case "4.0" => hive.v4_0
+      case "4.1" => hive.v4_1
+      case _ => throw QueryExecutionErrors.unsupportedHiveMetastoreVersionError(
         version, HiveUtils.HIVE_METASTORE_VERSION.key)
     }
   }
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
index 5506cf8dae073..2539f2539a323 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveFileFormat.scala
@@ -93,7 +93,7 @@ case class HiveFileFormat(fileSinkConf: FileSinkDesc)
     // Add table properties from storage handler to hadoopConf, so any custom storage
     // handler settings can be set to hadoopConf
     HiveTableUtil.configureJobPropertiesForStorageHandler(tableDesc, conf, false)
-    Utilities.copyTableJobPropertiesToConf(tableDesc, conf)
+    Utilities.copyTableJobPropertiesToConf(tableDesc, new JobConf(conf))
 
     // Avoid referencing the outer object.
     val fileSinkConfSer = fileSinkConf
@@ -140,9 +140,7 @@ class HiveOutputWriter(
   private def tableDesc = fileSinkConf.getTableInfo
 
   private val serializer = {
-    val serializer = tableDesc.getDeserializerClass.getConstructor().
-      newInstance().asInstanceOf[Serializer]
-    serializer.initialize(jobConf, tableDesc.getProperties)
+    val serializer = tableDesc.getDeserializer(jobConf).asInstanceOf[Serializer]
     serializer
   }
 
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationExec.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationExec.scala
index de2d15415837a..4535e723c9328 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationExec.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveScriptTransformationExec.scala
@@ -284,7 +284,7 @@ object HiveScriptIOSchema extends HiveInspectors {
 
     val properties = new Properties()
     properties.putAll(propsMap.asJava)
-    serde.initialize(null, properties)
+    serde.initialize(null, properties, null)
 
     serde
   }
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScanExec.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScanExec.scala
index 5e4ef82d798b8..dc6bb3f2576f6 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScanExec.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScanExec.scala
@@ -123,8 +123,7 @@ case class HiveTableScanExec(
 
     HiveShim.appendReadColumns(hiveConf, neededColumnIDs, neededColumnNames)
 
-    val deserializer = tableDesc.getDeserializerClass.getConstructor().newInstance()
-    deserializer.initialize(hiveConf, tableDesc.getProperties)
+    val deserializer = tableDesc.getDeserializer(hiveConf)
 
     // Specifies types and object inspectors of columns to be scanned.
     val structOI = ObjectInspectorUtils
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTempPath.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTempPath.scala
index 3864d1601504c..b4e94be7f597a 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTempPath.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTempPath.scala
@@ -149,7 +149,7 @@ class HiveTempPath(session: SparkSession, val hadoopConf: Configuration, path: P
     try {
       stagingDirForCreating.foreach { stagingDir =>
         val fs: FileSystem = stagingDir.getFileSystem(hadoopConf)
-        if (!FileUtils.mkdir(fs, stagingDir, true, hadoopConf)) {
+        if (!FileUtils.mkdir(fs, stagingDir, hadoopConf)) {
           throw SparkException.internalError(
             "Cannot create staging directory  '" + stagingDir.toString + "'")
         }
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala
index b5d3fb699d62e..06b32f9810b45 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/InsertIntoHiveTable.scala
@@ -146,9 +146,9 @@ case class InsertIntoHiveTable(
       if (numDynamicPartitions > 0) {
         if (overwrite && table.tableType == CatalogTableType.EXTERNAL) {
           val numWrittenParts = writtenParts.size
-          val maxDynamicPartitionsKey = HiveConf.ConfVars.DYNAMICPARTITIONMAXPARTS.varname
+          val maxDynamicPartitionsKey = HiveConf.ConfVars.DYNAMIC_PARTITION_MAX_PARTS.varname
           val maxDynamicPartitions = hadoopConf.getInt(maxDynamicPartitionsKey,
-            HiveConf.ConfVars.DYNAMICPARTITIONMAXPARTS.defaultIntVal)
+            HiveConf.ConfVars.DYNAMIC_PARTITION_MAX_PARTS.defaultIntVal)
           if (numWrittenParts > maxDynamicPartitions) {
             throw QueryExecutionErrors.writePartitionExceedConfigSizeWhenDynamicPartitionError(
               numWrittenParts, maxDynamicPartitions, maxDynamicPartitionsKey)
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
index ba37e5c176ded..4ff5fa980a4a5 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/orc/OrcFileFormat.scala
@@ -236,7 +236,7 @@ private[orc] class OrcSerializer(dataSchema: StructType, conf: Configuration)
     table.setProperty("columns.types", dataSchema.map(_.dataType.catalogString).mkString(":"))
 
     val serde = new OrcSerde
-    serde.initialize(conf, table)
+    serde.initialize(conf, table, null)
     serde
   }
 
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
index 86041d48cde75..e57177ab09de5 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveDDLSuite.scala
@@ -2776,10 +2776,13 @@ class HiveDDLSuite
             checkAnswer(spark.table("t"), Row(1))
             val maybeFile = path.listFiles().find(_.getName.startsWith("part"))
 
-            Utils.tryWithResource(getReader(maybeFile.head.getCanonicalPath)) { reader =>
+            val reader = getReader(maybeFile.head.getCanonicalPath)
+            try {
               assert(reader.getCompressionKind.name === OrcCompressionCodec.ZLIB.name())
               assert(reader.getCompressionSize == 1001)
               assert(reader.getRowIndexStride == 2002)
+            } finally {
+              reader.close()
             }
           }
         }
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala
index b82a69d526959..d6406e45caad0 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveQuerySuite.scala
@@ -1212,7 +1212,7 @@ class HiveQuerySuite extends HiveComparisonTest with SQLTestUtils with BeforeAnd
             .zip(parts)
             .map { case (k, v) =>
               if (v == "NULL") {
-                s"$k=${ConfVars.DEFAULTPARTITIONNAME.defaultStrVal}"
+                s"$k=${ConfVars.DEFAULT_PARTITION_NAME.defaultStrVal}"
               } else {
                 s"$k=$v"
               }
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala
index b7c466f3ed6a5..03ecad30deb8f 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/HiveUDFSuite.scala
@@ -856,7 +856,7 @@ class TestPair(x: Int, y: Int) extends Writable with Serializable {
 }
 
 class PairSerDe extends AbstractSerDe {
-  override def initialize(p1: Configuration, p2: Properties): Unit = {}
+  override def initialize(p1: Configuration, p2: Properties, p3: Properties): Unit = {}
 
   override def getObjectInspector: ObjectInspector = {
     ObjectInspectorFactory
diff --git a/sql/pipelines/pom.xml b/sql/pipelines/pom.xml
index c3b5ddf9dbaef..cb41b26844136 100644
--- a/sql/pipelines/pom.xml
+++ b/sql/pipelines/pom.xml
@@ -22,7 +22,7 @@
     <parent>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-parent_2.13</artifactId>
-        <version>4.1.3-SNAPSHOT</version>
+        <version>4.1.3.0</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
     <artifactId>spark-pipelines_2.13</artifactId>
diff --git a/streaming/pom.xml b/streaming/pom.xml
index c07d76714da4b..b65077824f0c0 100644
--- a/streaming/pom.xml
+++ b/streaming/pom.xml
@@ -21,7 +21,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 
diff --git a/tools/pom.xml b/tools/pom.xml
index 510b10e4de0db..86dd4c0812292 100644
--- a/tools/pom.xml
+++ b/tools/pom.xml
@@ -20,7 +20,7 @@
   <parent>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-parent_2.13</artifactId>
-    <version>4.1.3-SNAPSHOT</version>
+    <version>4.1.3.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>