Skip to content

Commit 4f24d7d

Browse files
davidyuan1223davidyuan1223
committed
Merge branch 'master' into test_remove_table_properties
# Conflicts: # extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/PaimonCatalogRangerSparkExtensionSuite.scala
2 parents 11d3773 + 4cab817 commit 4f24d7d

File tree

16 files changed

+165
-109
lines changed

16 files changed

+165
-109
lines changed

‎.github/workflows/master.yml

+2-2
Original file line numberDiff line numberDiff line change
@@ -444,8 +444,8 @@ jobs:
444444
# https://minikube.sigs.k8s.io/docs/handbook/pushing/#7-loading-directly-to-in-cluster-container-runtime
445445
minikube image load apache/kyuubi:latest
446446
# pre-install spark into minikube
447-
docker pull apache/spark:3.5.4
448-
minikube image load apache/spark:3.5.4
447+
docker pull apache/spark:3.5.5
448+
minikube image load apache/spark:3.5.5
449449
- name: kubectl pre-check
450450
run: |
451451
kubectl get nodes

‎bin/docker-image-tool.sh

+2-2
Original file line numberDiff line numberDiff line change
@@ -181,8 +181,8 @@ Examples:
181181
$0 -r docker.io/myrepo -t v1.8.1 build
182182
$0 -r docker.io/myrepo -t v1.8.1 push
183183
184-
- Build and push with tag "v1.8.1" and Spark-3.5.4 as base image to docker.io/myrepo
185-
$0 -r docker.io/myrepo -t v1.8.1 -b BASE_IMAGE=repo/spark:3.5.4 build
184+
- Build and push with tag "v1.8.1" and Spark-3.5.5 as base image to docker.io/myrepo
185+
$0 -r docker.io/myrepo -t v1.8.1 -b BASE_IMAGE=repo/spark:3.5.5 build
186186
$0 -r docker.io/myrepo -t v1.8.1 push
187187
188188
- Build and push for multiple archs to docker.io/myrepo

‎docs/deployment/kyuubi_on_kubernetes.md

+2-2
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,8 @@ Examples:
4242
$0 -r docker.io/myrepo -t v1.8.1 build
4343
$0 -r docker.io/myrepo -t v1.8.1 push
4444

45-
- Build and push with tag "v1.8.1" and Spark-3.5.4 as base image to docker.io/myrepo
46-
$0 -r docker.io/myrepo -t v1.8.1 -b BASE_IMAGE=repo/spark:3.5.4 build
45+
- Build and push with tag "v1.8.1" and Spark-3.5.5 as base image to docker.io/myrepo
46+
$0 -r docker.io/myrepo -t v1.8.1 -b BASE_IMAGE=repo/spark:3.5.5 build
4747
$0 -r docker.io/myrepo -t v1.8.1 push
4848

4949
- Build and push for multiple archs to docker.io/myrepo

‎docs/extensions/engines/spark/lineage.md

+1-1
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ Sometimes, it may be incompatible with other Spark distributions, then you may n
117117
For example,
118118

119119
```shell
120-
build/mvn clean package -pl :kyuubi-spark-lineage_2.12 -am -DskipTests -Dspark.version=3.5.4
120+
build/mvn clean package -pl :kyuubi-spark-lineage_2.12 -am -DskipTests -Dspark.version=3.5.5
121121
```
122122

123123
The available `spark.version`s are shown in the following table.

‎docs/monitor/events.md

-19
This file was deleted.

‎docs/monitor/index.rst

+1-2
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,8 @@ In this section, you will learn how to monitor Kyuubi with logging, metrics etc.
2020

2121
.. toctree::
2222
:maxdepth: 2
23-
:numbered: 3
2423
:glob:
2524

2625
logging
2726
metrics
28-
trouble_shooting
27+
troubleshooting

‎docs/monitor/logging.md

+3-4
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
- limitations under the License.
1616
-->
1717

18-
# Monitoring Kyuubi - Logging System
18+
# Logging
1919

2020
Kyuubi uses [Apache Log4j2](https://logging.apache.org/log4j/2.x/) for logging since version v1.5.0. For versions v1.4.1 and below, it uses [Apache Log4j](https://logging.apache.org).
2121

@@ -310,9 +310,8 @@ You will both get the final results and the corresponding operation logs telling
310310

311311
## Further Readings
312312

313-
- [Monitoring Kyuubi - Events System](events.md)
314-
- [Monitoring Kyuubi - Server Metrics](metrics.md)
315-
- [Trouble Shooting](trouble_shooting.md)
313+
- [Metrics](metrics.md)
314+
- [Troubleshooting](troubleshooting.md)
316315
- Spark Online Documentation
317316
- [Monitoring and Instrumentation](https://spark.apache.org/docs/latest/monitoring.html)
318317

‎docs/monitor/metrics.md

+1-1
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
- limitations under the License.
1616
-->
1717

18-
# Monitoring Kyuubi - Server Metrics
18+
# Metrics
1919

2020
Kyuubi has a configurable metrics system based on the [Dropwizard Metrics Library](https://metrics.dropwizard.io/).
2121
This allows users to report Kyuubi metrics to a variety of `kyuubi.metrics.reporters`.

‎docs/monitor/trouble_shooting.md ‎docs/monitor/troubleshooting.md

+1-1
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
- limitations under the License.
1616
-->
1717

18-
# Trouble Shooting
18+
# Troubleshooting
1919

2020
## Common Issues
2121

‎extensions/spark/kyuubi-spark-authz/src/test/scala/org/apache/kyuubi/plugin/spark/authz/ranger/PaimonCatalogRangerSparkExtensionSuite.scala

+54
Original file line numberDiff line numberDiff line change
@@ -170,6 +170,60 @@ class PaimonCatalogRangerSparkExtensionSuite extends RangerSparkExtensionSuite {
170170
}
171171
}
172172

173+
test("Rename Column Name") {
174+
withCleanTmpResources(Seq(
175+
(s"$catalogV2.$namespace1.$table1", "table"))) {
176+
val createTable = createTableSql(namespace1, table1)
177+
doAs(admin, sql(createTable))
178+
val renameColumnSql =
179+
s"""
180+
|ALTER TABLE $catalogV2.$namespace1.$table1
181+
|RENAME COLUMN name TO name1
182+
|""".stripMargin
183+
interceptEndsWith[AccessControlException] {
184+
doAs(someone, sql(renameColumnSql))
185+
}(s"does not have [alter] privilege on [$namespace1/$table1]")
186+
doAs(admin, sql(renameColumnSql))
187+
}
188+
}
189+
190+
test("Changing Column Position") {
191+
withCleanTmpResources(Seq(
192+
(s"$catalogV2.$namespace1.$table1", "table"))) {
193+
val createTableSql =
194+
s"""
195+
|CREATE TABLE IF NOT EXISTS $catalogV2.$namespace1.$table1
196+
|(id int, name string, a int, b int)
197+
|USING paimon
198+
|OPTIONS (
199+
| 'primary-key' = 'id'
200+
|)
201+
|""".stripMargin
202+
doAs(admin, sql(createTableSql))
203+
val changingColumnPositionToFirst =
204+
s"""
205+
|ALTER TABLE $catalogV2.$namespace1.$table1
206+
|ALTER COLUMN a FIRST
207+
|""".stripMargin
208+
209+
interceptEndsWith[AccessControlException] {
210+
doAs(someone, sql(changingColumnPositionToFirst))
211+
}(s"does not have [alter] privilege on [$namespace1/$table1]")
212+
doAs(admin, sql(changingColumnPositionToFirst))
213+
214+
val changingColumnPositionToAfter =
215+
s"""
216+
|ALTER TABLE $catalogV2.$namespace1.$table1
217+
|ALTER COLUMN a AFTER name
218+
|""".stripMargin
219+
220+
interceptEndsWith[AccessControlException] {
221+
doAs(someone, sql(changingColumnPositionToAfter))
222+
}(s"does not have [alter] privilege on [$namespace1/$table1]")
223+
doAs(admin, sql(changingColumnPositionToAfter))
224+
}
225+
}
226+
173227
test("REMOVING TBLEPROPERTIES") {
174228
withCleanTmpResources(Seq(
175229
(s"$catalogV2.$namespace1.$table1", "table"))) {

‎extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveConnectorUtils.scala

+63-63
Original file line numberDiff line numberDiff line change
@@ -19,36 +19,35 @@ package org.apache.kyuubi.spark.connector.hive
1919

2020
import java.lang.{Boolean => JBoolean, Long => JLong}
2121

22+
import scala.util.Try
23+
2224
import org.apache.hadoop.fs.{FileStatus, Path}
2325
import org.apache.hadoop.hive.ql.plan.{FileSinkDesc, TableDesc}
24-
import org.apache.spark.SPARK_VERSION
2526
import org.apache.spark.internal.Logging
2627
import org.apache.spark.sql.SparkSession
2728
import org.apache.spark.sql.catalyst.InternalRow
2829
import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogTablePartition}
2930
import org.apache.spark.sql.connector.catalog.TableChange
30-
import org.apache.spark.sql.connector.catalog.TableChange.{AddColumn, After, ColumnPosition, DeleteColumn, First, RenameColumn, UpdateColumnComment, UpdateColumnNullability, UpdateColumnPosition, UpdateColumnType}
31+
import org.apache.spark.sql.connector.catalog.TableChange._
3132
import org.apache.spark.sql.execution.command.CommandUtils
3233
import org.apache.spark.sql.execution.command.CommandUtils.{calculateMultipleLocationSizes, calculateSingleLocationSize}
3334
import org.apache.spark.sql.execution.datasources.{PartitionDirectory, PartitionedFile}
3435
import org.apache.spark.sql.hive.execution.HiveFileFormat
3536
import org.apache.spark.sql.internal.SQLConf
3637
import org.apache.spark.sql.types.{ArrayType, MapType, StructField, StructType}
3738

38-
import org.apache.kyuubi.spark.connector.common.SparkUtils.SPARK_RUNTIME_VERSION
3939
import org.apache.kyuubi.util.reflect.{DynClasses, DynConstructors, DynMethods}
4040
import org.apache.kyuubi.util.reflect.ReflectUtils.invokeAs
4141

4242
object HiveConnectorUtils extends Logging {
4343

44-
// SPARK-43186
45-
def getHiveFileFormat(fileSinkConf: FileSinkDesc): HiveFileFormat = {
46-
if (SPARK_RUNTIME_VERSION >= "3.5") {
44+
def getHiveFileFormat(fileSinkConf: FileSinkDesc): HiveFileFormat =
45+
Try { // SPARK-43186: 3.5.0
4746
DynConstructors.builder()
4847
.impl(classOf[HiveFileFormat], classOf[FileSinkDesc])
4948
.build[HiveFileFormat]()
5049
.newInstance(fileSinkConf)
51-
} else if (SPARK_RUNTIME_VERSION >= "3.3") {
50+
}.recover { case _: Exception =>
5251
val shimFileSinkDescClz = DynClasses.builder()
5352
.impl("org.apache.spark.sql.hive.HiveShim$ShimFileSinkDesc")
5453
.build()
@@ -67,34 +66,26 @@ object HiveConnectorUtils extends Logging {
6766
.impl(classOf[HiveFileFormat], shimFileSinkDescClz)
6867
.build[HiveFileFormat]()
6968
.newInstance(shimFileSinkDesc)
70-
} else {
71-
throw unsupportedSparkVersion()
72-
}
73-
}
69+
}.get
7470

75-
// SPARK-41970
76-
def partitionedFilePath(file: PartitionedFile): String = {
77-
if (SPARK_RUNTIME_VERSION >= "3.4") {
71+
def partitionedFilePath(file: PartitionedFile): String =
72+
Try { // SPARK-41970: 3.4.0
7873
invokeAs[String](file, "urlEncodedPath")
79-
} else if (SPARK_RUNTIME_VERSION >= "3.3") {
74+
}.recover { case _: Exception =>
8075
invokeAs[String](file, "filePath")
81-
} else {
82-
throw unsupportedSparkVersion()
83-
}
84-
}
76+
}.get
8577

8678
def splitFiles(
8779
sparkSession: SparkSession,
8880
file: AnyRef,
8981
filePath: Path,
90-
isSplitable: Boolean,
91-
maxSplitBytes: Long,
92-
partitionValues: InternalRow): Seq[PartitionedFile] = {
93-
94-
if (SPARK_RUNTIME_VERSION >= "4.0") { // SPARK-42821
82+
isSplitable: JBoolean,
83+
maxSplitBytes: JLong,
84+
partitionValues: InternalRow): Seq[PartitionedFile] =
85+
Try { // SPARK-42821: 4.0.0-preview2
9586
val fileStatusWithMetadataClz = DynClasses.builder()
9687
.impl("org.apache.spark.sql.execution.datasources.FileStatusWithMetadata")
97-
.build()
88+
.buildChecked()
9889
DynMethods
9990
.builder("splitFiles")
10091
.impl(
@@ -103,35 +94,58 @@ object HiveConnectorUtils extends Logging {
10394
classOf[Boolean],
10495
classOf[Long],
10596
classOf[InternalRow])
106-
.build()
107-
.invoke[Seq[PartitionedFile]](
97+
.buildChecked()
98+
.invokeChecked[Seq[PartitionedFile]](
10899
null,
109100
file,
110-
isSplitable.asInstanceOf[JBoolean],
111-
maxSplitBytes.asInstanceOf[JLong],
101+
isSplitable,
102+
maxSplitBytes,
112103
partitionValues)
113-
} else if (SPARK_RUNTIME_VERSION >= "3.5") { // SPARK-43039
104+
}.recover { case _: Exception => // SPARK-51185: Spark 3.5.5
114105
val fileStatusWithMetadataClz = DynClasses.builder()
115106
.impl("org.apache.spark.sql.execution.datasources.FileStatusWithMetadata")
116-
.build()
107+
.buildChecked()
117108
DynMethods
118109
.builder("splitFiles")
119110
.impl(
120111
"org.apache.spark.sql.execution.PartitionedFileUtil",
121112
classOf[SparkSession],
122113
fileStatusWithMetadataClz,
114+
classOf[Path],
123115
classOf[Boolean],
124116
classOf[Long],
125117
classOf[InternalRow])
126-
.build()
127-
.invoke[Seq[PartitionedFile]](
118+
.buildChecked()
119+
.invokeChecked[Seq[PartitionedFile]](
120+
null,
121+
sparkSession,
122+
file,
123+
filePath,
124+
isSplitable,
125+
maxSplitBytes,
126+
partitionValues)
127+
}.recover { case _: Exception => // SPARK-43039: 3.5.0
128+
val fileStatusWithMetadataClz = DynClasses.builder()
129+
.impl("org.apache.spark.sql.execution.datasources.FileStatusWithMetadata")
130+
.buildChecked()
131+
DynMethods
132+
.builder("splitFiles")
133+
.impl(
134+
"org.apache.spark.sql.execution.PartitionedFileUtil",
135+
classOf[SparkSession],
136+
fileStatusWithMetadataClz,
137+
classOf[Boolean],
138+
classOf[Long],
139+
classOf[InternalRow])
140+
.buildChecked()
141+
.invokeChecked[Seq[PartitionedFile]](
128142
null,
129143
sparkSession,
130144
file,
131-
isSplitable.asInstanceOf[JBoolean],
132-
maxSplitBytes.asInstanceOf[JLong],
145+
isSplitable,
146+
maxSplitBytes,
133147
partitionValues)
134-
} else if (SPARK_RUNTIME_VERSION >= "3.3") {
148+
}.recover { case _: Exception =>
135149
DynMethods
136150
.builder("splitFiles")
137151
.impl(
@@ -142,55 +156,41 @@ object HiveConnectorUtils extends Logging {
142156
classOf[Boolean],
143157
classOf[Long],
144158
classOf[InternalRow])
145-
.build()
146-
.invoke[Seq[PartitionedFile]](
159+
.buildChecked()
160+
.invokeChecked[Seq[PartitionedFile]](
147161
null,
148162
sparkSession,
149163
file,
150164
filePath,
151-
isSplitable.asInstanceOf[JBoolean],
152-
maxSplitBytes.asInstanceOf[JLong],
165+
isSplitable,
166+
maxSplitBytes,
153167
partitionValues)
154-
} else {
155-
throw unsupportedSparkVersion()
156-
}
157-
}
168+
}.get
158169

159-
def createPartitionDirectory(values: InternalRow, files: Seq[FileStatus]): PartitionDirectory = {
160-
if (SPARK_RUNTIME_VERSION >= "3.5") {
170+
def createPartitionDirectory(values: InternalRow, files: Seq[FileStatus]): PartitionDirectory =
171+
Try { // SPARK-43039: 3.5.0
161172
new DynMethods.Builder("apply")
162173
.impl(classOf[PartitionDirectory], classOf[InternalRow], classOf[Array[FileStatus]])
163174
.buildChecked()
164175
.asStatic()
165176
.invoke[PartitionDirectory](values, files.toArray)
166-
} else if (SPARK_RUNTIME_VERSION >= "3.3") {
177+
}.recover { case _: Exception =>
167178
new DynMethods.Builder("apply")
168179
.impl(classOf[PartitionDirectory], classOf[InternalRow], classOf[Seq[FileStatus]])
169180
.buildChecked()
170181
.asStatic()
171182
.invoke[PartitionDirectory](values, files)
172-
} else {
173-
throw unsupportedSparkVersion()
174-
}
175-
}
183+
}.get
176184

177-
def getPartitionFilePath(file: AnyRef): Path = {
178-
if (SPARK_RUNTIME_VERSION >= "3.5") {
185+
def getPartitionFilePath(file: AnyRef): Path =
186+
Try { // SPARK-43039: 3.5.0
179187
new DynMethods.Builder("getPath")
180188
.impl("org.apache.spark.sql.execution.datasources.FileStatusWithMetadata")
181189
.build()
182190
.invoke[Path](file)
183-
} else if (SPARK_RUNTIME_VERSION >= "3.3") {
191+
}.recover { case _: Exception =>
184192
file.asInstanceOf[FileStatus].getPath
185-
} else {
186-
throw unsupportedSparkVersion()
187-
}
188-
}
189-
190-
private def unsupportedSparkVersion(): KyuubiHiveConnectorException = {
191-
KyuubiHiveConnectorException(s"Spark version $SPARK_VERSION " +
192-
"is not supported by Kyuubi spark hive connector.")
193-
}
193+
}.get
194194

195195
def calculateTotalSize(
196196
spark: SparkSession,

0 commit comments

Comments
 (0)