peter-toth commented on code in PR #57069:
URL: https://github.com/apache/spark/pull/57069#discussion_r3535463721
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala:
##########
@@ -145,25 +145,37 @@ case class GetJsonObject(json: Expression, path:
Expression)
object GetJsonObject {
import PathInstruction._
- private[sql] def simpleNamedPath(path: UTF8String): Option[Seq[String]] = {
+ private[sql] sealed trait SimpleJsonPathSegment
+ private[sql] case class NamedPathSegment(name: String) extends
SimpleJsonPathSegment
+ private[sql] case class IndexedPathSegment(index: Long) extends
SimpleJsonPathSegment
+
+ private[sql] def simplePath(path: UTF8String):
Option[Seq[SimpleJsonPathSegment]] = {
try {
Option(path).flatMap(value =>
JsonPathParser.parse(value.toString)).flatMap { instructions =>
- val names = instructions.grouped(2).map {
- case List(Key, Named(fieldName)) => Some(fieldName)
+ val segments = instructions.grouped(2).map {
+ case List(Key, Named(fieldName)) => Some(NamedPathSegment(fieldName))
+ case List(Subscript, Index(index)) if index >= 0 =>
Some(IndexedPathSegment(index))
case _ => None
}.toSeq
- if (names.nonEmpty && names.forall(_.isDefined)) Some(names.flatten)
else None
+ if (segments.nonEmpty && segments.forall(_.isDefined))
Some(segments.flatten) else None
}
} catch {
// Numeric subscripts are parsed as Long and can overflow before the
parser returns None.
case _: NumberFormatException => None
}
}
+
+ private[sql] def simpleNamedPath(path: UTF8String): Option[Seq[String]] = {
Review Comment:
Seems like `simpleNamedPath` has no callers. This PR switched
`OptimizeCsvJsonExprs` and `MultiGetJsonObject` from `simpleNamedPath` to
`simplePath`, and a repo-wide search turns up only this definition -- so it
reads like a compatibility shim that nothing needs. Was it kept intentionally
as part of the `private[sql]` surface, or should it be removed as dead code?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala:
##########
@@ -44,76 +44,127 @@ import org.apache.spark.unsafe.types.UTF8String
object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
private def nameOfCorruptRecord = conf.columnNameOfCorruptRecord
+ private type SimpleJsonPath = Seq[GetJsonObject.SimpleJsonPathSegment]
+ private type SharedJsonCandidate = (GetJsonObject, SimpleJsonPath, String)
+ private type SharedJsonCandidateUnit = Seq[SharedJsonCandidate]
+ private type RequestedJsonPath = (SimpleJsonPath, String)
+ private type RequestedJsonPathUnit = Seq[RequestedJsonPath]
+
private case class SharedJsonFields(
json: Expression,
- paths: Seq[Seq[String]],
+ paths: Seq[SimpleJsonPath],
alias: Alias) {
- val ordinalMapping: Map[Seq[String], Int] = paths.zipWithIndex.toMap
+ val ordinalMapping: Map[SimpleJsonPath, Int] = paths.zipWithIndex.toMap
}
private final class SelectedJsonPathTrieNode {
var isTerminal: Boolean = false
var hasSelectedPathInSubtree: Boolean = false
- val children: mutable.HashMap[String, SelectedJsonPathTrieNode] =
mutable.HashMap.empty
+ val children: mutable.HashMap[GetJsonObject.SimpleJsonPathSegment,
SelectedJsonPathTrieNode] =
+ mutable.HashMap.empty
}
private final class SelectedJsonPathGroup {
val root = new SelectedJsonPathTrieNode
- val paths = mutable.ArrayBuffer.empty[(Seq[String], String)]
+ val paths = mutable.ArrayBuffer.empty[RequestedJsonPath]
- def tryAdd(candidate: (Seq[String], String)): Boolean = {
- val added = addIfNoPrefixConflict(root, candidate._1)
- if (added) {
- paths += candidate
+ def tryAdd(unit: RequestedJsonPathUnit): Boolean = {
+ val uniquePaths = mutable.LinkedHashMap.empty[SimpleJsonPath, String]
+ unit.foreach { case (path, fallbackPath) =>
+ uniquePaths.getOrElseUpdate(path, fallbackPath)
+ }
+ val newPaths = uniquePaths.iterator.filterNot { case (path, _) =>
+ containsExactPath(root, path)
+ }.toSeq
+ val conflictsWithGroup = newPaths.exists { case (path, _) =>
+ hasPrefixConflict(root, path)
+ }
+ val conflictsWithinUnit = newPaths.indices.exists { leftIndex =>
Review Comment:
Can `conflictsWithinUnit` ever be `true` with the units built today? Every
unit is either a single-path unit (from a standalone `GetJsonObject`) or the
two-element `Seq(firstNamed, firstIndexed)` pair from
`eligibleCoalesceBranches` -- and in that pair one path starts with a
`NamedPathSegment` and the other with an `IndexedPathSegment`, so neither can
strict-prefix the other and `isStrictPrefix` (used only here) is always
`false`. If that's right, it's currently unreachable.
--
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]
