LuciferYang commented on code in PR #58984:
URL: https://github.com/apache/spark/pull/58984#discussion_r4262017846
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala:
##########
@@ -1565,6 +1660,123 @@ private[client] class Shim_v4_0 extends Shim_v3_1 {
recordHiveCall()
renamePartitionMethod.invoke(hive, table, oldPartSpec, newPart,
writeIdInLoadTableOrPartition)
}
+
+ private[client] def normalizeLocation(root: String, relPath: String):
Option[URI] = {
+ val loc = if (relPath != null && relPath.nonEmpty) {
+ if (root != null) new Path(root, relPath.stripPrefix("/")).toString else
relPath
Review Comment:
`normalizeLocation` rebuilds the partition path with `new Path(root,
relPath.stripPrefix("/"))`, but HMS decides whether a partition sits under the
table with a plain string `startsWith(tablePath)` and sets `relativePath =
location.substring(tablePath.length())`. For a table at `/wh/t` and a partition
at `/wh/t_backup/p1=1`, HMS returns root `/wh/t` and relativePath
`_backup/p1=1`, and this code turns that into `/wh/t/_backup/p1=1`. I
reproduced it with the 4.0 client: both `INSERT INTO` and `INSERT OVERWRITE`
into that partition succeed, the new file lands in `t/_backup/p1=1`, and the
table doesn't see the new data.
Hive's own `PartitionSpecWithSharedSDProxy` concatenates `sd.getLocation() +
relativePath`, which is the exact inverse of the server's `substring`. Could we
do the same here? When the location is null, the old path used the table
location plus the partition name, and this path should do the same instead of
falling back to the table root. An empty relativePath needs to stay separate
from null, since it means the partition sits at the table location. The
`HiveClientImplSuite` assertions need to change with it, and a
sibling-directory case there would catch this.
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala:
##########
@@ -1565,6 +1660,123 @@ private[client] class Shim_v4_0 extends Shim_v3_1 {
recordHiveCall()
renamePartitionMethod.invoke(hive, table, oldPartSpec, newPart,
writeIdInLoadTableOrPartition)
}
+
+ private[client] def normalizeLocation(root: String, relPath: String):
Option[URI] = {
+ val loc = if (relPath != null && relPath.nonEmpty) {
+ if (root != null) new Path(root, relPath.stripPrefix("/")).toString else
relPath
+ } else {
+ root
+ }
+ Option(loc).map(CatalogUtils.stringToURI)
+ }
+
+ override def getPartitionSummaries(
+ hive: Hive,
+ table: Table,
+ partSpec: JMap[String, String],
+ mask: PartitionFieldMask): Seq[TablePartitionSummary] = {
+ if (!table.isPartitioned) {
+ return Seq.empty
+ }
+ val projSpec = clazzGetProjectionsSpec.getConstructor().newInstance()
+ val fieldList = if (mask.includeLocation) {
+ java.util.Arrays.asList("values", "sd.location")
+ } else {
+ java.util.Collections.singletonList("values")
+ }
+ setFieldListMethod.invoke(projSpec, fieldList)
+
+ val partColNames =
table.getPartCols.asScala.map(_.getName.toLowerCase(Locale.ROOT)).toSeq
+ val lowerPartSpec = partSpec.asScala.map { case (k, v) =>
+ k.toLowerCase(Locale.ROOT) -> v
+ }
+ val specKeys = lowerPartSpec.keySet
+ // Only push down BY_VALUES filter when the specified partition keys form
a strict prefix
+ // of partition columns; otherwise fetch projected partitions and filter
on client side.
+ val isPrefix = specKeys.nonEmpty && partColNames.take(specKeys.size).toSet
== specKeys
+ val filterSpec = if (isPrefix) {
+ val pvals = partColNames.take(specKeys.size).map(lowerPartSpec.apply)
+ val fs = clazzGetPartitionsFilterSpec.getConstructor().newInstance()
+ val byValues = clazzPartitionFilterMode.getEnumConstants
+ .find(_.toString == "BY_VALUES").get
+ setFilterModeMethod.invoke(fs, byValues)
+ setFiltersMethod.invoke(fs, pvals.asJava)
+ fs
+ } else {
+ // Hive 4.0 HMS ObjectStore unconditionally calls
filterSpec.isSetFilterMode(),
+ // so an empty GetPartitionsFilterSpec is required even when no filter
is applied.
+ clazzGetPartitionsFilterSpec.getConstructor().newInstance()
+ }
+
+ val req = clazzGetPartitionsRequest.getConstructor().newInstance()
+ setDbNameMethod.invoke(req, table.getDbName)
+ setTblNameMethod.invoke(req, table.getTableName)
+ Option(table.getTTable.getCatName).foreach { cat =>
+ setCatNameMethod.invoke(req, cat)
+ }
+ setProjectionSpecMethod.invoke(req, projSpec)
+ setFilterSpecMethod.invoke(req, filterSpec)
+
+ recordHiveCall()
+ val msc = getMSC(hive)
+ val response = getPartitionsWithSpecsMethod.invoke(msc, req)
Review Comment:
On the HMS side, `get_partitions_with_specs` only does `get_table_core` plus
the projection query. It skips what `get_partitions_ps_with_auth` does today:
the `hive.metastore.limit.partition.request` check,
`authorizeTableForPartitionMetadata`, and `filterPartitionsIfEnabled` both
server-side and in the client. The preceding `get_table` still fires the
table-level `PreReadTableEvent`, but on clusters that configure a partition
limit or a filter hook, INSERT and TRUNCATE would now bypass these checks
without notice.
I'd like to hear your take on adding an internal SQLConf that sends
`Shim_v4_0` back to `getPartitions`, plus a note in the description listing
which checks this RPC skips.
##########
sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveShim.scala:
##########
@@ -1565,6 +1660,123 @@ private[client] class Shim_v4_0 extends Shim_v3_1 {
recordHiveCall()
renamePartitionMethod.invoke(hive, table, oldPartSpec, newPart,
writeIdInLoadTableOrPartition)
}
+
+ private[client] def normalizeLocation(root: String, relPath: String):
Option[URI] = {
+ val loc = if (relPath != null && relPath.nonEmpty) {
+ if (root != null) new Path(root, relPath.stripPrefix("/")).toString else
relPath
+ } else {
+ root
+ }
+ Option(loc).map(CatalogUtils.stringToURI)
+ }
+
+ override def getPartitionSummaries(
+ hive: Hive,
+ table: Table,
+ partSpec: JMap[String, String],
+ mask: PartitionFieldMask): Seq[TablePartitionSummary] = {
+ if (!table.isPartitioned) {
+ return Seq.empty
+ }
+ val projSpec = clazzGetProjectionsSpec.getConstructor().newInstance()
+ val fieldList = if (mask.includeLocation) {
+ java.util.Arrays.asList("values", "sd.location")
+ } else {
+ java.util.Collections.singletonList("values")
+ }
+ setFieldListMethod.invoke(projSpec, fieldList)
+
+ val partColNames =
table.getPartCols.asScala.map(_.getName.toLowerCase(Locale.ROOT)).toSeq
+ val lowerPartSpec = partSpec.asScala.map { case (k, v) =>
+ k.toLowerCase(Locale.ROOT) -> v
+ }
+ val specKeys = lowerPartSpec.keySet
+ // Only push down BY_VALUES filter when the specified partition keys form
a strict prefix
+ // of partition columns; otherwise fetch projected partitions and filter
on client side.
+ val isPrefix = specKeys.nonEmpty && partColNames.take(specKeys.size).toSet
== specKeys
+ val filterSpec = if (isPrefix) {
+ val pvals = partColNames.take(specKeys.size).map(lowerPartSpec.apply)
Review Comment:
Only the prefix values go into BY_VALUES, and HMS's `makePartNameMatcher`
pads a short list by appending a wildcard with no `/` in between, so `p1=1`
also matches `p1=10/...` and `p1=100/...`. `HiveExternalCatalog` filters again,
so the result is right, but with integer partition keys this can fetch more
partitions than the old `getPartitions` call did: that path pads unspecified
columns with `""` via `MetaStoreUtils.getPvals`, giving `p1=1/p2=.*`.
Padding `filters` to the full partition-column list with `""` for the
missing ones should fix it, since `escapePathName` turns an empty value into
the wildcard. Both the directSQL and JDO branches handle it that way from 4.0
through 4.2, so non-prefix specs can be pushed down too, and `isPrefix` plus
the in-memory filter in the shim can go away. The second filter in
`HiveExternalCatalog` should stay.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
