[WIP] [SPARK-4198] [Core] Refactor some BlockManager internals

[aarondav]

Main goals of this PR:

• Break up doPut and doGetLocal into component methods which are individually more readable.
• Remove the weird "return bytes or value depending on parameter" in BlockManager.
• Move the block-related Serialization logic out of BlockManager.

Not quite finished and tested, just wanted to get what I have so far up.

[SparkQA]

Test build #22794 has started for PR 3065 at commit 791df20.

• This patch merges cleanly.

[SparkQA]

Test build #22794 has finished for PR 3065 at commit 791df20.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class BlockSerializer(conf: SparkConf, serializer: Serializer)
  • case class PutStatistics(size: Long, droppedBlocks: Seq[(BlockId, BlockStatus)])
  • trait PutResult
  • case class SuccessfulPut(statistics: PutStatistics) extends PutResult
  • case class OutOfSpaceFailure(
  • case class SparkStorage(level: StorageLevel)
  • case class OffHeapStorage(name: String)
  • // in some cases, such as when a class is enclosed in an object (in which case
  • abstract class UserDefinedType[UserType] extends DataType with Serializable
  • public abstract class UserDefinedType<UserType> extends DataType implements Serializable

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22794/
Test FAILed.

[SparkQA]

Test build #22801 has started for PR 3065 at commit 95971ae.

• This patch merges cleanly.

[SparkQA]

Test build #22801 has finished for PR 3065 at commit 95971ae.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class BlockSerializer(conf: SparkConf, serializer: Serializer)
  • case class PutStatistics(size: Long, droppedBlocks: Seq[(BlockId, BlockStatus)])
  • trait PutResult
  • case class SuccessfulPut(statistics: PutStatistics) extends PutResult
  • case class OutOfSpaceFailure(
  • case class SparkStorage(level: StorageLevel)
  • case class OffHeapStorage(name: String)

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22801/
Test FAILed.

[SparkQA]

Test build #22807 has started for PR 3065 at commit cb8bec0.

• This patch does not merge cleanly.

[SparkQA]

Test build #22807 has finished for PR 3065 at commit cb8bec0.

• This patch fails Spark unit tests.
• This patch does not merge cleanly.
• This patch adds the following public classes (experimental):
  • class BlockSerializer(conf: SparkConf, serializer: Serializer)
  • case class PutStatistics(size: Long, droppedBlocks: Seq[(BlockId, BlockStatus)])
  • trait PutResult
  • case class SuccessfulPut(statistics: PutStatistics) extends PutResult
  • case class OutOfSpaceFailure(
  • // in some cases, such as when a class is enclosed in an object (in which case
  • abstract class GenericStrategy[PhysicalPlan <: TreeNode[PhysicalPlan]] extends Logging
  • abstract class UserDefinedType[UserType] extends DataType with Serializable
  • public abstract class UserDefinedType<UserType> extends DataType implements Serializable
  • trait RunnableCommand extends logical.Command
  • case class ExecutedCommand(cmd: RunnableCommand) extends SparkPlan
  • protected case class Keyword(str: String)
  • sys.error(s"Failed to load class for data source: $provider")
  • case class EqualTo(attribute: String, value: Any) extends Filter
  • case class GreaterThan(attribute: String, value: Any) extends Filter
  • case class GreaterThanOrEqual(attribute: String, value: Any) extends Filter
  • case class LessThan(attribute: String, value: Any) extends Filter
  • case class LessThanOrEqual(attribute: String, value: Any) extends Filter
  • trait RelationProvider
  • abstract class BaseRelation
  • abstract class TableScan extends BaseRelation
  • abstract class PrunedScan extends BaseRelation
  • abstract class PrunedFilteredScan extends BaseRelation

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22807/
Test FAILed.

[tdas]

I think these functions can be renamed to serializeStream, deserializeStream, etc. That is, remove data from the names.

[tdas]

In fact, the names can be made more imperative (similar to wrapForCompression): serializeToStream, deserializeFromStream, etc.

[tdas]

Is this PR still relevant? Otherwise mind closing this and opening it again later when it is updated and ready for review?

[aarondav]

It is still relevant in that this is a much-needed refactor, but it is not ready for review, so I will close it.