{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/21859","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/21859/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/21859/comments","events_url":"https://api.github.com/repos/apache/spark/issues/21859/events","html_url":"https://github.com/apache/spark/pull/21859","id":343985785,"node_id":"MDExOlB1bGxSZXF1ZXN0MjAzNDg2OTY0","number":21859,"title":"[SPARK-24900][SQL]Speed up sort when the dataset is small","user":{"login":"sddyljsx","id":3488126,"node_id":"MDQ6VXNlcjM0ODgxMjY=","avatar_url":"https://avatars2.githubusercontent.com/u/3488126?v=4","gravatar_id":"","url":"https://api.github.com/users/sddyljsx","html_url":"https://github.com/sddyljsx","followers_url":"https://api.github.com/users/sddyljsx/followers","following_url":"https://api.github.com/users/sddyljsx/following{/other_user}","gists_url":"https://api.github.com/users/sddyljsx/gists{/gist_id}","starred_url":"https://api.github.com/users/sddyljsx/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sddyljsx/subscriptions","organizations_url":"https://api.github.com/users/sddyljsx/orgs","repos_url":"https://api.github.com/users/sddyljsx/repos","events_url":"https://api.github.com/users/sddyljsx/events{/privacy}","received_events_url":"https://api.github.com/users/sddyljsx/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":31,"created_at":"2018-07-24T10:55:28Z","updated_at":"2018-08-21T02:59:26Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/21859","html_url":"https://github.com/apache/spark/pull/21859","diff_url":"https://github.com/apache/spark/pull/21859.diff","patch_url":"https://github.com/apache/spark/pull/21859.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nwhen running the sql like 'select * from order where order_status = 4 order by order_id'. The filescan and filter will be executed twice, it may take a long time. If the final dataset is small, and the sample data covers all the data, there is no need to do so.\r\n\r\n## How was this patch tested?\r\n\r\nUnit Test added in SQLQuerySuite.\r\n\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/414534002","html_url":"https://github.com/apache/spark/pull/21859#issuecomment-414534002","issue_url":"https://api.github.com/repos/apache/spark/issues/21859","id":414534002,"node_id":"MDEyOklzc3VlQ29tbWVudDQxNDUzNDAwMg==","user":{"login":"sddyljsx","id":3488126,"node_id":"MDQ6VXNlcjM0ODgxMjY=","avatar_url":"https://avatars2.githubusercontent.com/u/3488126?v=4","gravatar_id":"","url":"https://api.github.com/users/sddyljsx","html_url":"https://github.com/sddyljsx","followers_url":"https://api.github.com/users/sddyljsx/followers","following_url":"https://api.github.com/users/sddyljsx/following{/other_user}","gists_url":"https://api.github.com/users/sddyljsx/gists{/gist_id}","starred_url":"https://api.github.com/users/sddyljsx/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sddyljsx/subscriptions","organizations_url":"https://api.github.com/users/sddyljsx/orgs","repos_url":"https://api.github.com/users/sddyljsx/repos","events_url":"https://api.github.com/users/sddyljsx/events{/privacy}","received_events_url":"https://api.github.com/users/sddyljsx/received_events","type":"User","site_admin":false},"created_at":"2018-08-21T02:59:26Z","updated_at":"2018-08-21T02:59:26Z","author_association":"CONTRIBUTOR","body":"I read the source code again. \r\nThe RangePartitioner[K, V] in ShuffleExchangeExec is an instance of RangePartitioner[InternalRow, Null]. RangePartitioner only sample K for getting the rangeBounds. So We can get the InternalRow when doing sample.\r\nAfter getting the RangePartitioner, the ShuffleExchangeExec will map the InternalRow to [partitionId, InternalRow] for shuffle (the RangePartitioner generates the partitionId).\r\nThe shuffle won't use the RangePartitioner, it will use PartitionIdPassthrough instead.\r\nIn other words, the ShuffleWriter won't know the RangePartitioner's existence.\r\n\r\n```\r\nval rddWithPartitionIds: RDD[Product2[Int, InternalRow]] = newRdd.mapPartitionsInternal { iter =>\r\n          val getPartitionKey = getPartitionKeyExtractor()\r\n          val mutablePair = new MutablePair[Int, InternalRow]()\r\n          iter.map { row => mutablePair.update(part.getPartition(getPartitionKey(row)), row) }\r\n        }\r\n\r\n val dependency =\r\n      new ShuffleDependency[Int, InternalRow, InternalRow](\r\n        rddWithPartitionIds,\r\n        new PartitionIdPassthrough(part.numPartitions),\r\n        serializer)\r\n\r\nprivate class PartitionIdPassthrough(override val numPartitions: Int) extends Partitioner {\r\n  override def getPartition(key: Any): Int = key.asInstanceOf[Int]\r\n}\r\n```\r\n\r\nThe optimization will parallelize the cached InternalRow to the newRdd instead of getting it again.\r\n\r\nBut in other places, like rdd's sortByKey\r\n\r\n```\r\ndef sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length)\r\n      : RDD[(K, V)] = self.withScope\r\n  {\r\n    val part = new RangePartitioner(numPartitions, self, ascending)\r\n    new ShuffledRDD[K, V, V](self, part)\r\n      .setKeyOrdering(if (ascending) ordering else ordering.reverse)\r\n  }\r\n// getDependencies function in ShuffledRDD\r\noverride def getDependencies: Seq[Dependency[_]] = {\r\n    val serializer = userSpecifiedSerializer.getOrElse {\r\n      val serializerManager = SparkEnv.get.serializerManager\r\n      if (mapSideCombine) {\r\n        serializerManager.getSerializer(implicitly[ClassTag[K]], implicitly[ClassTag[C]])\r\n      } else {\r\n        serializerManager.getSerializer(implicitly[ClassTag[K]], implicitly[ClassTag[V]])\r\n      }\r\n    }\r\n    List(new ShuffleDependency(prev, part, serializer, keyOrdering, aggregator, mapSideCombine))\r\n  }\r\n\r\n```\r\nThe rdd is [K, V], and the shuffle uses RangePartitioner directly.  But we can only get K when doing sample. so we can't restore the rdd using the cache.\r\n\r\nThey work in two different ways.\r\n\r\nSo the optimization only works in Spark Sql's ShuffleExchangeExec by now.\r\n\r\n'The ShuffleWriter should treat RangePartitioner specially and consume the sampled data in RangePartitioner instead of the input iterator.' This idea is good, maybe we can cache both the K and V when doing sample. I will have a try."}}
{"actor":{"display_login":"sddyljsx"}}