Reformat files with scalafmt (#473)
This commit is contained in:
Chungmin Lee
GitHub
Родитель
38bd40019e
Коммит
896f734d4d
|
|
@ -1,3 +1,5 @@
|
|||
version = 2.6.3
|
||||
|
||||
# The following configs are taken from https://github.com/apache/spark/blob/master/dev/.scalafmt.conf
|
||||
align = none
|
||||
align.openParenDefnSite = false
|
||||
|
|
@ -11,4 +13,4 @@ docstrings = JavaDoc
|
|||
maxColumn = 98
|
||||
|
||||
# The following are specific to Hyperspace.
|
||||
importSelectors = singleLine
|
||||
importSelectors = singleLine
|
||||
|
|
|
|||
|
|
@ -32,10 +32,11 @@ import com.microsoft.hyperspace.util.JavaConverters._
|
|||
|
||||
object IcebergShims {
|
||||
|
||||
def isIcebergRelation(plan: LogicalPlan): Boolean = plan match {
|
||||
case DataSourceV2Relation(_: IcebergSource, _, _, _, _) => true
|
||||
case _ => false
|
||||
}
|
||||
def isIcebergRelation(plan: LogicalPlan): Boolean =
|
||||
plan match {
|
||||
case DataSourceV2Relation(_: IcebergSource, _, _, _, _) => true
|
||||
case _ => false
|
||||
}
|
||||
|
||||
def loadIcebergTable(spark: SparkSession, plan: LogicalPlan): (Table, Option[Long]) = {
|
||||
val conf = spark.sessionState.newHadoopConf()
|
||||
|
|
|
|||
|
|
@ -28,11 +28,12 @@ object IcebergShims {
|
|||
|
||||
// In Spark 3, the V2ScanRelationPushdown rule can convert DataSourceV2Relation into
|
||||
// DataSourceV2ScanRelation.
|
||||
def isIcebergRelation(plan: LogicalPlan): Boolean = plan match {
|
||||
case DataSourceV2Relation(_: SparkTable, _, _, _, _) => true
|
||||
case DataSourceV2ScanRelation(_: SparkTable, _, _) => true
|
||||
case _ => false
|
||||
}
|
||||
def isIcebergRelation(plan: LogicalPlan): Boolean =
|
||||
plan match {
|
||||
case DataSourceV2Relation(_: SparkTable, _, _, _, _) => true
|
||||
case DataSourceV2ScanRelation(_: SparkTable, _, _) => true
|
||||
case _ => false
|
||||
}
|
||||
|
||||
def loadIcebergTable(spark: SparkSession, plan: LogicalPlan): (Table, Option[Long]) =
|
||||
plan match {
|
||||
|
|
|
|||
|
|
@ -26,11 +26,12 @@ import org.apache.spark.util.hyperspace.Utils
|
|||
|
||||
object IcebergShims {
|
||||
|
||||
def isIcebergRelation(plan: LogicalPlan): Boolean = plan match {
|
||||
case DataSourceV2Relation(_: SparkTable, _, _, _, _) => true
|
||||
case DataSourceV2ScanRelation(DataSourceV2Relation(_: SparkTable, _, _, _, _), _, _) => true
|
||||
case _ => false
|
||||
}
|
||||
def isIcebergRelation(plan: LogicalPlan): Boolean =
|
||||
plan match {
|
||||
case DataSourceV2Relation(_: SparkTable, _, _, _, _) => true
|
||||
case DataSourceV2ScanRelation(DataSourceV2Relation(_: SparkTable, _, _, _, _), _, _) => true
|
||||
case _ => false
|
||||
}
|
||||
|
||||
def loadIcebergTable(spark: SparkSession, plan: LogicalPlan): (Table, Option[Long]) =
|
||||
plan match {
|
||||
|
|
|
|||
|
|
@ -78,25 +78,29 @@ trait ScalaObjectMapper {
|
|||
if (isArray(clazz)) {
|
||||
val typeArguments = m.typeArguments.map(constructType(_)).toArray
|
||||
if (typeArguments.length != 1) {
|
||||
throw new IllegalArgumentException("Need exactly 1 type parameter for array like types ("+clazz.getName+")")
|
||||
throw new IllegalArgumentException(
|
||||
"Need exactly 1 type parameter for array like types (" + clazz.getName + ")")
|
||||
}
|
||||
getTypeFactory.constructArrayType(typeArguments(0))
|
||||
} else if (isMapLike(clazz)) {
|
||||
val typeArguments = m.typeArguments.map(constructType(_)).toArray
|
||||
if (typeArguments.length != 2) {
|
||||
throw new IllegalArgumentException("Need exactly 2 type parameters for map like types ("+clazz.getName+")")
|
||||
throw new IllegalArgumentException(
|
||||
"Need exactly 2 type parameters for map like types (" + clazz.getName + ")")
|
||||
}
|
||||
getTypeFactory.constructMapLikeType(clazz, typeArguments(0), typeArguments(1))
|
||||
} else if (isReference(clazz)) { // Option is a subclss of IterableOnce, so check it first
|
||||
val typeArguments = m.typeArguments.map(constructType(_)).toArray
|
||||
if (typeArguments.length != 1) {
|
||||
throw new IllegalArgumentException("Need exactly 1 type parameter for reference types ("+clazz.getName+")")
|
||||
throw new IllegalArgumentException(
|
||||
"Need exactly 1 type parameter for reference types (" + clazz.getName + ")")
|
||||
}
|
||||
getTypeFactory.constructReferenceType(clazz, typeArguments(0))
|
||||
} else if (isCollectionLike(clazz)) {
|
||||
val typeArguments = m.typeArguments.map(constructType(_)).toArray
|
||||
if (typeArguments.length != 1) {
|
||||
throw new IllegalArgumentException("Need exactly 1 type parameter for collection like types ("+clazz.getName+")")
|
||||
throw new IllegalArgumentException(
|
||||
"Need exactly 1 type parameter for collection like types (" + clazz.getName + ")")
|
||||
}
|
||||
getTypeFactory.constructCollectionLikeType(clazz, typeArguments(0))
|
||||
} else {
|
||||
|
|
@ -355,7 +359,7 @@ trait ScalaObjectMapper {
|
|||
c.isArray
|
||||
}
|
||||
|
||||
private val MAP = classOf[collection.Map[_,_]]
|
||||
private val MAP = classOf[collection.Map[_, _]]
|
||||
private def isMapLike(c: Class[_]): Boolean = {
|
||||
MAP.isAssignableFrom(c)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -156,8 +156,8 @@ class Hyperspace(spark: SparkSession) {
|
|||
* @param redirectFunc optional function to redirect output of explain.
|
||||
* @param verbose Flag to enable verbose mode.
|
||||
*/
|
||||
def explain(df: DataFrame, verbose: Boolean = false)(
|
||||
implicit redirectFunc: String => Unit = print): Unit = {
|
||||
def explain(df: DataFrame, verbose: Boolean = false)(implicit
|
||||
redirectFunc: String => Unit = print): Unit = {
|
||||
redirectFunc(PlanAnalyzer.explainString(df, spark, indexManager.indexes, verbose))
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -61,12 +61,7 @@ class CancelAction(final override protected val logManager: IndexLogManager) ext
|
|||
*/
|
||||
final override def op(): Unit = {}
|
||||
|
||||
final override protected def event(
|
||||
appInfo: AppInfo,
|
||||
message: String): HyperspaceEvent = {
|
||||
CancelActionEvent(
|
||||
appInfo,
|
||||
logEntry.asInstanceOf[IndexLogEntry],
|
||||
message)
|
||||
final override protected def event(appInfo: AppInfo, message: String): HyperspaceEvent = {
|
||||
CancelActionEvent(appInfo, logEntry.asInstanceOf[IndexLogEntry], message)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -66,8 +66,7 @@ class CreateAction(
|
|||
}
|
||||
|
||||
// TODO: Temporarily block creating indexes using nested columns until it's fully supported.
|
||||
if (!(HyperspaceConf.nestedColumnEnabled(spark) || resolvedColumns.get.forall(
|
||||
!_.isNested))) {
|
||||
if (!(HyperspaceConf.nestedColumnEnabled(spark) || resolvedColumns.get.forall(!_.isNested))) {
|
||||
throw HyperspaceException("Hyperspace does not support nested columns yet.")
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -72,12 +72,8 @@ class RefreshIncrementalAction(
|
|||
} else {
|
||||
None
|
||||
}
|
||||
updatedIndex = Some(
|
||||
previousIndexLogEntry.derivedDataset.refreshIncremental(
|
||||
this,
|
||||
appendedSourceData,
|
||||
deletedFiles,
|
||||
previousIndexLogEntry.content))
|
||||
updatedIndex = Some(previousIndexLogEntry.derivedDataset
|
||||
.refreshIncremental(this, appendedSourceData, deletedFiles, previousIndexLogEntry.content))
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -146,10 +146,11 @@ case class CoveringIndex(
|
|||
indexData)
|
||||
}
|
||||
|
||||
override def equals(o: Any): Boolean = o match {
|
||||
case that: CoveringIndex => comparedData == that.comparedData
|
||||
case _ => false
|
||||
}
|
||||
override def equals(o: Any): Boolean =
|
||||
o match {
|
||||
case that: CoveringIndex => comparedData == that.comparedData
|
||||
case _ => false
|
||||
}
|
||||
|
||||
override def hashCode: Int = {
|
||||
comparedData.hashCode
|
||||
|
|
|
|||
|
|
@ -28,11 +28,7 @@ import org.apache.spark.sql.DataFrame
|
|||
*
|
||||
* The framework manages various types of indexes through this interface.
|
||||
*/
|
||||
@JsonTypeInfo(
|
||||
use = JsonTypeInfo.Id.CLASS,
|
||||
include = JsonTypeInfo.As.PROPERTY,
|
||||
property = "type"
|
||||
)
|
||||
@JsonTypeInfo(use = JsonTypeInfo.Id.CLASS, include = JsonTypeInfo.As.PROPERTY, property = "type")
|
||||
trait Index {
|
||||
|
||||
/**
|
||||
|
|
@ -126,10 +122,10 @@ trait Index {
|
|||
* index if no update is needed
|
||||
*/
|
||||
def refreshIncremental(
|
||||
ctx: IndexerContext,
|
||||
appendedSourceData: Option[DataFrame],
|
||||
deletedSourceDataFiles: Seq[FileInfo],
|
||||
indexContent: Content): Index
|
||||
ctx: IndexerContext,
|
||||
appendedSourceData: Option[DataFrame],
|
||||
deletedSourceDataFiles: Seq[FileInfo],
|
||||
indexContent: Content): Index
|
||||
|
||||
/**
|
||||
* Indexes the source data and returns an updated index and index data.
|
||||
|
|
|
|||
|
|
@ -50,7 +50,9 @@ case class Content(root: Directory, fingerprint: NoOpFingerprint = NoOpFingerpri
|
|||
rec(
|
||||
new Path(root.name),
|
||||
root,
|
||||
(f, prefix) =>
|
||||
(
|
||||
f,
|
||||
prefix) =>
|
||||
FileInfo(new Path(prefix, f.name).toString, f.size, f.modifiedTime, f.id)).toSet
|
||||
}
|
||||
|
||||
|
|
@ -101,9 +103,7 @@ object Content {
|
|||
* @param fileIdTracker FileIdTracker to keep mapping of file properties to assigned file ids.
|
||||
* @return Content object with Directory tree from leaf files.
|
||||
*/
|
||||
def fromLeafFiles(
|
||||
files: Seq[FileStatus],
|
||||
fileIdTracker: FileIdTracker): Option[Content] = {
|
||||
def fromLeafFiles(files: Seq[FileStatus], fileIdTracker: FileIdTracker): Option[Content] = {
|
||||
if (files.nonEmpty) {
|
||||
Some(Content(Directory.fromLeafFiles(files, fileIdTracker)))
|
||||
} else {
|
||||
|
|
@ -224,12 +224,10 @@ object Directory {
|
|||
* @param files List of leaf files.
|
||||
* @param fileIdTracker FileIdTracker to keep mapping of file properties to assigned file ids.
|
||||
* Note: If a new leaf file is discovered, the input fileIdTracker gets
|
||||
* updated by adding it to the files it is tracking.
|
||||
* updated by adding it to the files it is tracking.
|
||||
* @return Content object with Directory tree from leaf files.
|
||||
*/
|
||||
def fromLeafFiles(
|
||||
files: Seq[FileStatus],
|
||||
fileIdTracker: FileIdTracker): Directory = {
|
||||
def fromLeafFiles(files: Seq[FileStatus], fileIdTracker: FileIdTracker): Directory = {
|
||||
require(
|
||||
files.nonEmpty,
|
||||
s"Empty files list found while creating a ${Directory.getClass.getName}.")
|
||||
|
|
@ -308,13 +306,14 @@ object Directory {
|
|||
// id is a unique identifier generated by Hyperspace, for each unique combination of
|
||||
// file's name, size and modifiedTime.
|
||||
case class FileInfo(name: String, size: Long, modifiedTime: Long, id: Long) {
|
||||
override def equals(o: Any): Boolean = o match {
|
||||
case that: FileInfo =>
|
||||
name.equals(that.name) &&
|
||||
size.equals(that.size) &&
|
||||
modifiedTime.equals(that.modifiedTime)
|
||||
case _ => false
|
||||
}
|
||||
override def equals(o: Any): Boolean =
|
||||
o match {
|
||||
case that: FileInfo =>
|
||||
name.equals(that.name) &&
|
||||
size.equals(that.size) &&
|
||||
modifiedTime.equals(that.modifiedTime)
|
||||
case _ => false
|
||||
}
|
||||
|
||||
override def hashCode(): Int = {
|
||||
name.hashCode + size.hashCode + modifiedTime.hashCode
|
||||
|
|
@ -349,9 +348,7 @@ object LogicalPlanFingerprint {
|
|||
* @param appendedFiles Appended files.
|
||||
* @param deletedFiles Deleted files.
|
||||
*/
|
||||
case class Update(
|
||||
appendedFiles: Option[Content] = None,
|
||||
deletedFiles: Option[Content] = None)
|
||||
case class Update(appendedFiles: Option[Content] = None, deletedFiles: Option[Content] = None)
|
||||
|
||||
// IndexLogEntry-specific Hdfs that represents the source data.
|
||||
case class Hdfs(properties: Hdfs.Properties) {
|
||||
|
|
@ -467,34 +464,28 @@ case class IndexLogEntry(
|
|||
def toFileStatus(f: FileInfo) = {
|
||||
new FileStatus(f.size, false, 0, 1, f.modifiedTime, new Path(f.name))
|
||||
}
|
||||
copy(
|
||||
source = source.copy(
|
||||
plan = source.plan.copy(
|
||||
properties = source.plan.properties.copy(
|
||||
fingerprint = latestFingerprint,
|
||||
relations = Seq(
|
||||
relations.head.copy(
|
||||
data = relations.head.data.copy(
|
||||
properties = relations.head.data.properties.copy(
|
||||
update = Some(
|
||||
Update(
|
||||
appendedFiles =
|
||||
Content.fromLeafFiles(appended.map(toFileStatus), fileIdTracker),
|
||||
deletedFiles =
|
||||
Content.fromLeafFiles(deleted.map(toFileStatus), fileIdTracker)))))))))))
|
||||
copy(source = source.copy(plan = source.plan.copy(properties = source.plan.properties.copy(
|
||||
fingerprint = latestFingerprint,
|
||||
relations = Seq(
|
||||
relations.head.copy(data = relations.head.data.copy(properties =
|
||||
relations.head.data.properties.copy(update = Some(Update(
|
||||
appendedFiles = Content.fromLeafFiles(appended.map(toFileStatus), fileIdTracker),
|
||||
deletedFiles =
|
||||
Content.fromLeafFiles(deleted.map(toFileStatus), fileIdTracker)))))))))))
|
||||
}
|
||||
|
||||
override def equals(o: Any): Boolean = o match {
|
||||
case that: IndexLogEntry =>
|
||||
name.equals(that.name) &&
|
||||
derivedDataset.equals(that.derivedDataset) &&
|
||||
signature.equals(that.signature) &&
|
||||
content.root.equals(that.content.root) &&
|
||||
source.equals(that.source) &&
|
||||
properties.equals(that.properties) &&
|
||||
state.equals(that.state)
|
||||
case _ => false
|
||||
}
|
||||
override def equals(o: Any): Boolean =
|
||||
o match {
|
||||
case that: IndexLogEntry =>
|
||||
name.equals(that.name) &&
|
||||
derivedDataset.equals(that.derivedDataset) &&
|
||||
signature.equals(that.signature) &&
|
||||
content.root.equals(that.content.root) &&
|
||||
source.equals(that.source) &&
|
||||
properties.equals(that.properties) &&
|
||||
state.equals(that.state)
|
||||
case _ => false
|
||||
}
|
||||
|
||||
def indexedColumns: Seq[String] = derivedDataset.indexedColumns
|
||||
|
||||
|
|
@ -506,8 +497,9 @@ case class IndexLogEntry(
|
|||
|
||||
def hasParquetAsSourceFormat: Boolean = {
|
||||
relations.head.fileFormat.equals("parquet") ||
|
||||
derivedDataset.properties.getOrElse(
|
||||
IndexConstants.HAS_PARQUET_AS_SOURCE_FORMAT_PROPERTY, "false").toBoolean
|
||||
derivedDataset.properties
|
||||
.getOrElse(IndexConstants.HAS_PARQUET_AS_SOURCE_FORMAT_PROPERTY, "false")
|
||||
.toBoolean
|
||||
}
|
||||
|
||||
@JsonIgnore
|
||||
|
|
@ -594,8 +586,7 @@ object IndexLogEntry {
|
|||
derivedDataset,
|
||||
content,
|
||||
source,
|
||||
properties + ((IndexConstants.HYPERSPACE_VERSION_PROPERTY, BuildInfo.version))
|
||||
)
|
||||
properties + ((IndexConstants.HYPERSPACE_VERSION_PROPERTY, BuildInfo.version)))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -608,10 +599,10 @@ class FileIdTracker {
|
|||
// Combination of file properties, used as key, to identify a
|
||||
// unique file for which an id is generated.
|
||||
type key = (
|
||||
String, // Full path.
|
||||
String, // Full path.
|
||||
Long, // Size.
|
||||
Long // Modified time.
|
||||
)
|
||||
Long // Modified time.
|
||||
)
|
||||
private val fileToIdMap: mutable.HashMap[key, Long] = mutable.HashMap()
|
||||
|
||||
def getMaxFileId: Long = maxId
|
||||
|
|
@ -636,8 +627,7 @@ class FileIdTracker {
|
|||
setSizeHint(files.size)
|
||||
files.foreach { f =>
|
||||
if (f.id == IndexConstants.UNKNOWN_FILE_ID) {
|
||||
throw HyperspaceException(
|
||||
s"Cannot add file info with unknown id. (file: ${f.name}).")
|
||||
throw HyperspaceException(s"Cannot add file info with unknown id. (file: ${f.name}).")
|
||||
}
|
||||
|
||||
val key = (f.name, f.size, f.modifiedTime)
|
||||
|
|
@ -665,8 +655,7 @@ class FileIdTracker {
|
|||
*/
|
||||
def addFile(file: FileStatus): Long = {
|
||||
fileToIdMap.getOrElseUpdate(
|
||||
(file.getPath.toString, file.getLen, file.getModificationTime),
|
||||
{
|
||||
(file.getPath.toString, file.getLen, file.getModificationTime), {
|
||||
maxId += 1
|
||||
maxId
|
||||
})
|
||||
|
|
|
|||
|
|
@ -47,12 +47,12 @@ object IndexLogEntryTags {
|
|||
// INMEMORYFILEINDEX_HYBRID_SCAN stores InMemoryFileIndex including index data files and also
|
||||
// appended files for Hybrid Scan.
|
||||
val INMEMORYFILEINDEX_HYBRID_SCAN: IndexLogEntryTag[InMemoryFileIndex] =
|
||||
IndexLogEntryTag[InMemoryFileIndex]("inMemoryFileIndexHybridScan")
|
||||
IndexLogEntryTag[InMemoryFileIndex]("inMemoryFileIndexHybridScan")
|
||||
|
||||
// INMEMORYFILEINDEX_HYBRID_SCAN_APPENDED stores InMemoryFileIndex including only appended files
|
||||
// for Hybrid Scan.
|
||||
val INMEMORYFILEINDEX_HYBRID_SCAN_APPENDED: IndexLogEntryTag[InMemoryFileIndex] =
|
||||
IndexLogEntryTag[InMemoryFileIndex]("inMemoryFileIndexHybridScanAppended")
|
||||
IndexLogEntryTag[InMemoryFileIndex]("inMemoryFileIndexHybridScanAppended")
|
||||
|
||||
// FILTER_REASONS stores reason strings for disqualification.
|
||||
val FILTER_REASONS: IndexLogEntryTag[Seq[String]] =
|
||||
|
|
|
|||
|
|
@ -54,7 +54,7 @@ trait IndexLogManager {
|
|||
}
|
||||
|
||||
class IndexLogManagerImpl(indexPath: Path, hadoopConfiguration: Configuration = new Configuration)
|
||||
extends IndexLogManager
|
||||
extends IndexLogManager
|
||||
with Logging {
|
||||
// Use FileContext instead of FileSystem for atomic renames?
|
||||
private lazy val fs: FileSystem = indexPath.getFileSystem(hadoopConfiguration)
|
||||
|
|
@ -102,8 +102,9 @@ class IndexLogManagerImpl(indexPath: Path, hadoopConfiguration: Configuration =
|
|||
if (entry.exists(e => Constants.STABLE_STATES.contains(e.state))) {
|
||||
return entry
|
||||
}
|
||||
if (entry.exists(e => e.state.equals(Constants.States.CREATING)
|
||||
|| e.state.equals(Constants.States.VACUUMING))) {
|
||||
if (entry.exists(e =>
|
||||
e.state.equals(Constants.States.CREATING)
|
||||
|| e.state.equals(Constants.States.VACUUMING))) {
|
||||
// Do not consider unrelated logs before creating or vacuuming state.
|
||||
return None
|
||||
}
|
||||
|
|
|
|||
|
|
@ -55,12 +55,8 @@ private[hyperspace] case class IndexStatistics(
|
|||
additionalStats: Map[String, String])
|
||||
|
||||
private[hyperspace] object IndexStatistics {
|
||||
val INDEX_SUMMARY_COLUMNS: Seq[String] = Seq(
|
||||
"name",
|
||||
"indexedColumns",
|
||||
"indexLocation",
|
||||
"state",
|
||||
"additionalStats")
|
||||
val INDEX_SUMMARY_COLUMNS: Seq[String] =
|
||||
Seq("name", "indexedColumns", "indexLocation", "state", "additionalStats")
|
||||
|
||||
/**
|
||||
* Create IndexStatistics instance for a given IndexLogEntry.
|
||||
|
|
@ -146,7 +142,7 @@ private[hyperspace] object IndexStatistics {
|
|||
var root = entry.content.root
|
||||
var prefix = entry.content.root.name
|
||||
while (root.subDirs.size == 1 &&
|
||||
!root.subDirs.head.name.startsWith(IndexConstants.INDEX_VERSION_DIRECTORY_PREFIX)) {
|
||||
!root.subDirs.head.name.startsWith(IndexConstants.INDEX_VERSION_DIRECTORY_PREFIX)) {
|
||||
prefix += s"${root.subDirs.head.name}/"
|
||||
root = root.subDirs.head
|
||||
}
|
||||
|
|
|
|||
|
|
@ -26,9 +26,10 @@ import com.microsoft.hyperspace.index.plans.logical.BucketUnion
|
|||
* to [[BucketUnionExec]] (Spark Plan)
|
||||
*/
|
||||
private[hyperspace] object BucketUnionStrategy extends SparkStrategy {
|
||||
override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
|
||||
case p: BucketUnion =>
|
||||
BucketUnionExec(p.children.map(planLater), p.bucketSpec) :: Nil
|
||||
case _ => Nil
|
||||
}
|
||||
override def apply(plan: LogicalPlan): Seq[SparkPlan] =
|
||||
plan match {
|
||||
case p: BucketUnion =>
|
||||
BucketUnionExec(p.children.map(planLater), p.bucketSpec) :: Nil
|
||||
case _ => Nil
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -140,7 +140,7 @@ object PlanAnalyzer {
|
|||
val usedPaths = new ListBuffer[String]
|
||||
sparkPlan.foreach {
|
||||
case ExtractFileSourceScanExecRelation(
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _)) =>
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _)) =>
|
||||
usedPaths += location.rootPaths.head.getParent.toString
|
||||
case other =>
|
||||
other.subqueries.foreach { subQuery =>
|
||||
|
|
|
|||
|
|
@ -62,7 +62,7 @@ private[hyperspace] case class BucketUnion(children: Seq[LogicalPlan], bucketSpe
|
|||
// compare the data types with the first child
|
||||
child.output.zip(children.head.output).forall {
|
||||
case (l, r) => l.dataType.equals(r.dataType)
|
||||
})
|
||||
})
|
||||
children.length > 1 && childrenResolved && allChildrenCompatible
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -32,8 +32,8 @@ import com.microsoft.hyperspace.telemetry.HyperspaceEventLogging
|
|||
* Collect candidate indexes for each source plan.
|
||||
*/
|
||||
object CandidateIndexCollector extends ActiveSparkSession {
|
||||
private val sourceFilters
|
||||
: Seq[SourcePlanIndexFilter] = ColumnSchemaFilter :: FileSignatureFilter :: Nil
|
||||
private val sourceFilters: Seq[SourcePlanIndexFilter] =
|
||||
ColumnSchemaFilter :: FileSignatureFilter :: Nil
|
||||
|
||||
private def initializePlanToIndexes(
|
||||
plan: LogicalPlan,
|
||||
|
|
@ -56,9 +56,11 @@ object CandidateIndexCollector extends ActiveSparkSession {
|
|||
val planToIndexes = initializePlanToIndexes(plan, allIndexes)
|
||||
planToIndexes.flatMap {
|
||||
case (node, allIndexes) =>
|
||||
Some(node, sourceFilters.foldLeft(allIndexes) { (indexes, filter) =>
|
||||
filter(node, indexes)
|
||||
}).filter(_._2.nonEmpty)
|
||||
Some(
|
||||
node,
|
||||
sourceFilters.foldLeft(allIndexes) { (indexes, filter) =>
|
||||
filter(node, indexes)
|
||||
}).filter(_._2.nonEmpty)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -128,7 +128,7 @@ object FilterRankFilter extends IndexRankFilter {
|
|||
plan: LogicalPlan,
|
||||
applicableIndexes: PlanToIndexesMap): PlanToSelectedIndexMap = {
|
||||
if (applicableIndexes.isEmpty || applicableIndexes.size != 1
|
||||
|| applicableIndexes.head._2.isEmpty) {
|
||||
|| applicableIndexes.head._2.isEmpty) {
|
||||
Map.empty
|
||||
} else {
|
||||
val relation = RuleUtils.getRelation(spark, plan).get
|
||||
|
|
|
|||
|
|
@ -44,8 +44,8 @@ trait IndexFilter extends ActiveSparkSession {
|
|||
index: IndexLogEntry,
|
||||
reasonString: => String): Unit = {
|
||||
if (!condition && index
|
||||
.getTagValue(IndexLogEntryTags.FILTER_REASONS_ENABLED)
|
||||
.getOrElse(false)) {
|
||||
.getTagValue(IndexLogEntryTags.FILTER_REASONS_ENABLED)
|
||||
.getOrElse(false)) {
|
||||
val prevReason =
|
||||
index.getTagValue(plan, IndexLogEntryTags.FILTER_REASONS).getOrElse(Nil)
|
||||
index.setTagValue(plan, IndexLogEntryTags.FILTER_REASONS, prevReason :+ reasonString)
|
||||
|
|
|
|||
|
|
@ -52,13 +52,15 @@ object JoinPlanNodeFilter extends QueryPlanIndexFilter {
|
|||
val left = RuleUtils.getRelation(spark, l)
|
||||
val right = RuleUtils.getRelation(spark, r)
|
||||
|
||||
if (!(left.isDefined && right.isDefined && !RuleUtils.isIndexApplied(left.get) && !RuleUtils
|
||||
.isIndexApplied(right.get))) {
|
||||
if (!(left.isDefined && right.isDefined && !RuleUtils.isIndexApplied(
|
||||
left.get) && !RuleUtils
|
||||
.isIndexApplied(right.get))) {
|
||||
return Map.empty
|
||||
}
|
||||
|
||||
val leftAndRightIndexes = candidateIndexes.getOrElse(left.get.plan, Nil) ++ candidateIndexes
|
||||
.getOrElse(right.get.plan, Nil)
|
||||
val leftAndRightIndexes =
|
||||
candidateIndexes.getOrElse(left.get.plan, Nil) ++ candidateIndexes
|
||||
.getOrElse(right.get.plan, Nil)
|
||||
|
||||
val joinConditionCond = withFilterReasonTag(
|
||||
plan,
|
||||
|
|
@ -160,17 +162,17 @@ object JoinAttributeFilter extends QueryPlanIndexFilter {
|
|||
}
|
||||
|
||||
if (withFilterReasonTag(
|
||||
plan,
|
||||
candidateIndexes.head._2 ++ candidateIndexes.last._2,
|
||||
"Each join condition column should come from " +
|
||||
"relations directly and attributes from left plan must exclusively have " +
|
||||
"one-to-one mapping with attributes from right plan. " +
|
||||
"E.g. join(A = B and A = D) is not eligible.") {
|
||||
ensureAttributeRequirements(
|
||||
JoinIndexRule.leftRelation.get,
|
||||
JoinIndexRule.rightRelation.get,
|
||||
JoinIndexRule.joinCondition.get)
|
||||
}) {
|
||||
plan,
|
||||
candidateIndexes.head._2 ++ candidateIndexes.last._2,
|
||||
"Each join condition column should come from " +
|
||||
"relations directly and attributes from left plan must exclusively have " +
|
||||
"one-to-one mapping with attributes from right plan. " +
|
||||
"E.g. join(A = B and A = D) is not eligible.") {
|
||||
ensureAttributeRequirements(
|
||||
JoinIndexRule.leftRelation.get,
|
||||
JoinIndexRule.rightRelation.get,
|
||||
JoinIndexRule.joinCondition.get)
|
||||
}) {
|
||||
candidateIndexes
|
||||
} else {
|
||||
Map.empty
|
||||
|
|
@ -206,7 +208,6 @@ object JoinAttributeFilter extends QueryPlanIndexFilter {
|
|||
* E.g. (A = B and A = D) is not supported. A maps with both B and D. There isn't a one-to-one
|
||||
* mapping.
|
||||
*
|
||||
*
|
||||
* Background knowledge:
|
||||
* An alias in a query plan is represented as [[Alias]] at the time of
|
||||
* its creation. Unnecessary aliases get resolved and removed during query analysis phase by
|
||||
|
|
@ -329,13 +330,13 @@ object JoinColumnFilter extends QueryPlanIndexFilter {
|
|||
val rRequiredAllCols = resolve(spark, allRequiredCols(r), rBaseAttrs).get
|
||||
|
||||
if (withFilterReasonTag(
|
||||
plan,
|
||||
candidateIndexes.head._2 ++ candidateIndexes.last._2,
|
||||
"Invalid query plan.") {
|
||||
// Make sure required indexed columns are subset of all required columns.
|
||||
resolve(spark, lRequiredIndexedCols, lRequiredAllCols).isDefined &&
|
||||
resolve(spark, rRequiredIndexedCols, rRequiredAllCols).isDefined
|
||||
}) {
|
||||
plan,
|
||||
candidateIndexes.head._2 ++ candidateIndexes.last._2,
|
||||
"Invalid query plan.") {
|
||||
// Make sure required indexed columns are subset of all required columns.
|
||||
resolve(spark, lRequiredIndexedCols, lRequiredAllCols).isDefined &&
|
||||
resolve(spark, rRequiredIndexedCols, rRequiredAllCols).isDefined
|
||||
}) {
|
||||
val lIndexes =
|
||||
getUsableIndexes(
|
||||
plan,
|
||||
|
|
@ -350,13 +351,13 @@ object JoinColumnFilter extends QueryPlanIndexFilter {
|
|||
rRequiredAllCols)
|
||||
|
||||
if (withFilterReasonTag(
|
||||
plan,
|
||||
candidateIndexes.head._2 ++ candidateIndexes.last._2,
|
||||
"No available indexes for left subplan.")(lIndexes.nonEmpty) &&
|
||||
withFilterReasonTag(
|
||||
plan,
|
||||
candidateIndexes.head._2 ++ candidateIndexes.last._2,
|
||||
"No available indexes for right subplan.")(rIndexes.nonEmpty)) {
|
||||
plan,
|
||||
candidateIndexes.head._2 ++ candidateIndexes.last._2,
|
||||
"No available indexes for left subplan.")(lIndexes.nonEmpty) &&
|
||||
withFilterReasonTag(
|
||||
plan,
|
||||
candidateIndexes.head._2 ++ candidateIndexes.last._2,
|
||||
"No available indexes for right subplan.")(rIndexes.nonEmpty)) {
|
||||
Map(leftRelation.plan -> lIndexes, rightRelation.plan -> rIndexes)
|
||||
} else {
|
||||
Map.empty
|
||||
|
|
|
|||
|
|
@ -218,7 +218,7 @@ object RuleUtils {
|
|||
|
||||
val filesToRead = {
|
||||
if (useBucketSpec || !index.hasParquetAsSourceFormat || filesDeleted.nonEmpty ||
|
||||
relation.partitionSchema.nonEmpty) {
|
||||
relation.partitionSchema.nonEmpty) {
|
||||
// Since the index data is in "parquet" format, we cannot read source files
|
||||
// in formats other than "parquet" using one FileScan node as the operator requires
|
||||
// files in one homogenous format. To address this, we need to read the appended
|
||||
|
|
@ -388,16 +388,17 @@ object RuleUtils {
|
|||
// Extract top level plan including all required columns for shuffle in its output.
|
||||
object ExtractTopLevelPlanForShuffle {
|
||||
type returnType = (LogicalPlan, Seq[Option[Attribute]], Boolean)
|
||||
def unapply(plan: LogicalPlan): Option[returnType] = plan match {
|
||||
case p @ Project(_, Filter(_, LogicalRelation(_: HadoopFsRelation, _, _, _))) =>
|
||||
Some(p, getIndexedAttrs(p, bucketSpec.bucketColumnNames), true)
|
||||
case p @ Project(_, LogicalRelation(_: HadoopFsRelation, _, _, _)) =>
|
||||
Some(p, getIndexedAttrs(p, bucketSpec.bucketColumnNames), true)
|
||||
case f @ Filter(_, LogicalRelation(_: HadoopFsRelation, _, _, _)) =>
|
||||
Some(f, getIndexedAttrs(f, bucketSpec.bucketColumnNames), false)
|
||||
case r @ LogicalRelation(_: HadoopFsRelation, _, _, _) =>
|
||||
Some(r, getIndexedAttrs(r, bucketSpec.bucketColumnNames), false)
|
||||
}
|
||||
def unapply(plan: LogicalPlan): Option[returnType] =
|
||||
plan match {
|
||||
case p @ Project(_, Filter(_, LogicalRelation(_: HadoopFsRelation, _, _, _))) =>
|
||||
Some(p, getIndexedAttrs(p, bucketSpec.bucketColumnNames), true)
|
||||
case p @ Project(_, LogicalRelation(_: HadoopFsRelation, _, _, _)) =>
|
||||
Some(p, getIndexedAttrs(p, bucketSpec.bucketColumnNames), true)
|
||||
case f @ Filter(_, LogicalRelation(_: HadoopFsRelation, _, _, _)) =>
|
||||
Some(f, getIndexedAttrs(f, bucketSpec.bucketColumnNames), false)
|
||||
case r @ LogicalRelation(_: HadoopFsRelation, _, _, _) =>
|
||||
Some(r, getIndexedAttrs(r, bucketSpec.bucketColumnNames), false)
|
||||
}
|
||||
|
||||
private def getIndexedAttrs(
|
||||
plan: LogicalPlan,
|
||||
|
|
|
|||
|
|
@ -37,11 +37,13 @@ import com.microsoft.hyperspace.util.{CacheWithTransform, HyperspaceConf}
|
|||
*/
|
||||
class FileBasedSourceProviderManager(spark: SparkSession) {
|
||||
private val sourceProviders: CacheWithTransform[String, Seq[FileBasedSourceProvider]] =
|
||||
new CacheWithTransform[String, Seq[FileBasedSourceProvider]]({ () =>
|
||||
HyperspaceConf.fileBasedSourceBuilders(spark)
|
||||
}, { builderClassNames =>
|
||||
buildProviders(builderClassNames)
|
||||
})
|
||||
new CacheWithTransform[String, Seq[FileBasedSourceProvider]](
|
||||
{ () =>
|
||||
HyperspaceConf.fileBasedSourceBuilders(spark)
|
||||
},
|
||||
{ builderClassNames =>
|
||||
buildProviders(builderClassNames)
|
||||
})
|
||||
|
||||
/**
|
||||
* Returns true if the given logical plan is a supported relation. If all of the registered
|
||||
|
|
|
|||
|
|
@ -42,14 +42,15 @@ class DefaultFileBasedRelation(spark: SparkSession, override val plan: LogicalRe
|
|||
/**
|
||||
* Computes the signature of the current relation.
|
||||
*/
|
||||
override def signature: String = plan.relation match {
|
||||
case HadoopFsRelation(location: PartitioningAwareFileIndex, _, _, _, _, _) =>
|
||||
val result = filesFromIndex(location).sortBy(_.getPath.toString).foldLeft("") {
|
||||
(acc: String, f: FileStatus) =>
|
||||
HashingUtils.md5Hex(acc + fingerprint(f))
|
||||
}
|
||||
result
|
||||
}
|
||||
override def signature: String =
|
||||
plan.relation match {
|
||||
case HadoopFsRelation(location: PartitioningAwareFileIndex, _, _, _, _, _) =>
|
||||
val result = filesFromIndex(location).sortBy(_.getPath.toString).foldLeft("") {
|
||||
(acc: String, f: FileStatus) =>
|
||||
HashingUtils.md5Hex(acc + fingerprint(f))
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
/**
|
||||
* FileStatus list for all source files that the current relation references to.
|
||||
|
|
@ -62,28 +63,30 @@ class DefaultFileBasedRelation(spark: SparkSession, override val plan: LogicalRe
|
|||
/**
|
||||
* The partition schema of the current relation.
|
||||
*/
|
||||
override def partitionSchema: StructType = plan.relation match {
|
||||
case HadoopFsRelation(location: FileIndex, _, _, _, _, _) =>
|
||||
location.partitionSchema
|
||||
}
|
||||
override def partitionSchema: StructType =
|
||||
plan.relation match {
|
||||
case HadoopFsRelation(location: FileIndex, _, _, _, _, _) =>
|
||||
location.partitionSchema
|
||||
}
|
||||
|
||||
/**
|
||||
* The optional partition base path of the current relation.
|
||||
*/
|
||||
override def partitionBasePath: Option[String] = plan.relation match {
|
||||
case HadoopFsRelation(p: PartitioningAwareFileIndex, _, _, _, _, _)
|
||||
if p.partitionSpec.partitions.nonEmpty =>
|
||||
// For example, we could have the following in PartitionSpec:
|
||||
// - partition columns = "col1", "col2"
|
||||
// - partitions: "/path/col1=1/col2=1", "/path/col1=1/col2=2", etc.
|
||||
// , and going up the same number of directory levels as the number of partition columns
|
||||
// will compute the base path. Note that PartitionSpec.partitions will always contain
|
||||
// all the partitions in the path, so "partitions.head" is taken as an initial value.
|
||||
val basePath = p.partitionSpec.partitionColumns
|
||||
.foldLeft(p.partitionSpec.partitions.head.path)((path, _) => path.getParent)
|
||||
Some(basePath.toString)
|
||||
case _ => None
|
||||
}
|
||||
override def partitionBasePath: Option[String] =
|
||||
plan.relation match {
|
||||
case HadoopFsRelation(p: PartitioningAwareFileIndex, _, _, _, _, _)
|
||||
if p.partitionSpec.partitions.nonEmpty =>
|
||||
// For example, we could have the following in PartitionSpec:
|
||||
// - partition columns = "col1", "col2"
|
||||
// - partitions: "/path/col1=1/col2=1", "/path/col1=1/col2=2", etc.
|
||||
// , and going up the same number of directory levels as the number of partition columns
|
||||
// will compute the base path. Note that PartitionSpec.partitions will always contain
|
||||
// all the partitions in the path, so "partitions.head" is taken as an initial value.
|
||||
val basePath = p.partitionSpec.partitionColumns
|
||||
.foldLeft(p.partitionSpec.partitions.head.path)((path, _) => path.getParent)
|
||||
Some(basePath.toString)
|
||||
case _ => None
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates [[HadoopFsRelation]] based on the current relation.
|
||||
|
|
@ -93,10 +96,11 @@ class DefaultFileBasedRelation(spark: SparkSession, override val plan: LogicalRe
|
|||
override def createHadoopFsRelation(
|
||||
location: FileIndex,
|
||||
dataSchema: StructType,
|
||||
options: Map[String, String]): HadoopFsRelation = plan.relation match {
|
||||
case h: HadoopFsRelation =>
|
||||
h.copy(location = location, dataSchema = dataSchema, options = options)(spark)
|
||||
}
|
||||
options: Map[String, String]): HadoopFsRelation =
|
||||
plan.relation match {
|
||||
case h: HadoopFsRelation =>
|
||||
h.copy(location = location, dataSchema = dataSchema, options = options)(spark)
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates [[LogicalRelation]] based on the current relation.
|
||||
|
|
@ -118,12 +122,12 @@ class DefaultFileBasedRelation(spark: SparkSession, override val plan: LogicalRe
|
|||
override def createRelationMetadata(fileIdTracker: FileIdTracker): Relation = {
|
||||
plan.relation match {
|
||||
case HadoopFsRelation(
|
||||
location: PartitioningAwareFileIndex,
|
||||
_,
|
||||
dataSchema,
|
||||
_,
|
||||
fileFormat,
|
||||
options) =>
|
||||
location: PartitioningAwareFileIndex,
|
||||
_,
|
||||
dataSchema,
|
||||
_,
|
||||
fileFormat,
|
||||
options) =>
|
||||
val files = filesFromIndex(location)
|
||||
// Note that source files are currently fingerprinted when the optimized plan is
|
||||
// fingerprinted by LogicalPlanFingerprint.
|
||||
|
|
@ -215,10 +219,11 @@ class DefaultFileBasedRelation(spark: SparkSession, override val plan: LogicalRe
|
|||
*
|
||||
* @return True if source files of the current relation are parquet.
|
||||
*/
|
||||
def hasParquetAsSourceFormat: Boolean = plan.relation match {
|
||||
case h: HadoopFsRelation =>
|
||||
h.fileFormat.asInstanceOf[DataSourceRegister].shortName.equals("parquet")
|
||||
}
|
||||
def hasParquetAsSourceFormat: Boolean =
|
||||
plan.relation match {
|
||||
case h: HadoopFsRelation =>
|
||||
h.fileFormat.asInstanceOf[DataSourceRegister].shortName.equals("parquet")
|
||||
}
|
||||
|
||||
/**
|
||||
* For [[DefaultFileBasedRelation]], each file path should be in this format:
|
||||
|
|
|
|||
|
|
@ -36,11 +36,13 @@ import com.microsoft.hyperspace.util.{CacheWithTransform, HyperspaceConf}
|
|||
*/
|
||||
class DefaultFileBasedSource(private val spark: SparkSession) extends FileBasedSourceProvider {
|
||||
private val supportedFormats: CacheWithTransform[String, Set[String]] =
|
||||
new CacheWithTransform[String, Set[String]]({ () =>
|
||||
HyperspaceConf.supportedFileFormatsForDefaultFileBasedSource(spark)
|
||||
}, { formats =>
|
||||
formats.toLowerCase(Locale.ROOT).split(",").map(_.trim).toSet
|
||||
})
|
||||
new CacheWithTransform[String, Set[String]](
|
||||
{ () =>
|
||||
HyperspaceConf.supportedFileFormatsForDefaultFileBasedSource(spark)
|
||||
},
|
||||
{ formats =>
|
||||
formats.toLowerCase(Locale.ROOT).split(",").map(_.trim).toSet
|
||||
})
|
||||
|
||||
/**
|
||||
* Returns true if the given [[FileFormat]] is supported, false otherwise.
|
||||
|
|
@ -71,15 +73,16 @@ class DefaultFileBasedSource(private val spark: SparkSession) extends FileBasedS
|
|||
* @param plan Logical plan to check if it's supported.
|
||||
* @return Some(true) if the given plan is a supported relation, otherwise None.
|
||||
*/
|
||||
def isSupportedRelation(plan: LogicalPlan): Option[Boolean] = plan match {
|
||||
case LogicalRelation(
|
||||
HadoopFsRelation(_: PartitioningAwareFileIndex, _, _, _, fileFormat, _),
|
||||
_,
|
||||
_,
|
||||
_) if isSupportedFileFormat(fileFormat) =>
|
||||
Some(true)
|
||||
case _ => None
|
||||
}
|
||||
def isSupportedRelation(plan: LogicalPlan): Option[Boolean] =
|
||||
plan match {
|
||||
case LogicalRelation(
|
||||
HadoopFsRelation(_: PartitioningAwareFileIndex, _, _, _, fileFormat, _),
|
||||
_,
|
||||
_,
|
||||
_) if isSupportedFileFormat(fileFormat) =>
|
||||
Some(true)
|
||||
case _ => None
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the [[FileBasedRelation]] that wraps the given logical plan if the given
|
||||
|
|
|
|||
|
|
@ -45,11 +45,12 @@ class DeltaLakeFileBasedSource(private val spark: SparkSession) extends FileBase
|
|||
* @param plan Logical plan to check if it's supported.
|
||||
* @return Some(true) if the given plan is a supported relation, otherwise None.
|
||||
*/
|
||||
def isSupportedRelation(plan: LogicalPlan): Option[Boolean] = plan match {
|
||||
case LogicalRelation(HadoopFsRelation(_: TahoeLogFileIndex, _, _, _, _, _), _, _, _) =>
|
||||
Some(true)
|
||||
case _ => None
|
||||
}
|
||||
def isSupportedRelation(plan: LogicalPlan): Option[Boolean] =
|
||||
plan match {
|
||||
case LogicalRelation(HadoopFsRelation(_: TahoeLogFileIndex, _, _, _, _, _), _, _, _) =>
|
||||
Some(true)
|
||||
case _ => None
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the [[FileBasedRelation]] that wraps the given logical plan if the given
|
||||
|
|
|
|||
|
|
@ -37,10 +37,11 @@ class DeltaLakeRelation(spark: SparkSession, override val plan: LogicalRelation)
|
|||
/**
|
||||
* Computes the signature of the current relation.
|
||||
*/
|
||||
override def signature: String = plan.relation match {
|
||||
case HadoopFsRelation(location: TahoeLogFileIndex, _, _, _, _, _) =>
|
||||
location.tableVersion + location.path.toString
|
||||
}
|
||||
override def signature: String =
|
||||
plan.relation match {
|
||||
case HadoopFsRelation(location: TahoeLogFileIndex, _, _, _, _, _) =>
|
||||
location.tableVersion + location.path.toString
|
||||
}
|
||||
|
||||
/**
|
||||
* All the files that the current relation references to.
|
||||
|
|
@ -57,11 +58,12 @@ class DeltaLakeRelation(spark: SparkSession, override val plan: LogicalRelation)
|
|||
/**
|
||||
* The optional partition base path of the current relation.
|
||||
*/
|
||||
override def partitionBasePath: Option[String] = plan.relation match {
|
||||
case HadoopFsRelation(t: TahoeLogFileIndex, _, _, _, _, _) if t.partitionSchema.nonEmpty =>
|
||||
Some(t.path.toString)
|
||||
case _ => None
|
||||
}
|
||||
override def partitionBasePath: Option[String] =
|
||||
plan.relation match {
|
||||
case HadoopFsRelation(t: TahoeLogFileIndex, _, _, _, _, _) if t.partitionSchema.nonEmpty =>
|
||||
Some(t.path.toString)
|
||||
case _ => None
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates [[Relation]] for IndexLogEntry using the current relation.
|
||||
|
|
@ -179,8 +181,8 @@ class DeltaLakeRelation(spark: SparkSession, override val plan: LogicalRelation)
|
|||
// TODO: Support time travel utilizing Hybrid Scan append-only.
|
||||
// See https://github.com/microsoft/hyperspace/issues/408.
|
||||
if (!(HyperspaceConf.hybridScanEnabled(spark) &&
|
||||
HyperspaceConf.hybridScanDeleteEnabled(spark) &&
|
||||
index.derivedDataset.canHandleDeletedFiles)) {
|
||||
HyperspaceConf.hybridScanDeleteEnabled(spark) &&
|
||||
index.derivedDataset.canHandleDeletedFiles)) {
|
||||
return index
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -47,9 +47,12 @@ class DeltaLakeRelationMetadata(metadata: Relation) extends FileBasedRelationMet
|
|||
val deltaVerHistory = metadata.options.get("versionAsOf").map { deltaVersion =>
|
||||
val newVersionMapping = s"$indexVersion:$deltaVersion"
|
||||
DeltaLakeConstants.DELTA_VERSION_HISTORY_PROPERTY ->
|
||||
properties.get(DeltaLakeConstants.DELTA_VERSION_HISTORY_PROPERTY).map { prop =>
|
||||
s"$prop,$newVersionMapping"
|
||||
}.getOrElse(newVersionMapping)
|
||||
properties
|
||||
.get(DeltaLakeConstants.DELTA_VERSION_HISTORY_PROPERTY)
|
||||
.map { prop =>
|
||||
s"$prop,$newVersionMapping"
|
||||
}
|
||||
.getOrElse(newVersionMapping)
|
||||
}
|
||||
properties ++ deltaVerHistory
|
||||
}
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ package object hyperspace {
|
|||
def isHyperspaceEnabled(): Boolean = {
|
||||
val experimentalMethods = sparkSession.sessionState.experimentalMethods
|
||||
experimentalMethods.extraOptimizations.contains(ApplyHyperspace) &&
|
||||
experimentalMethods.extraStrategies.contains(BucketUnionStrategy)
|
||||
experimentalMethods.extraStrategies.contains(BucketUnionStrategy)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -114,7 +114,7 @@ case class CancelActionEvent(appInfo: AppInfo, index: IndexLogEntry, message: St
|
|||
* @param message Message about event.
|
||||
*/
|
||||
case class RefreshIncrementalActionEvent(appInfo: AppInfo, index: IndexLogEntry, message: String)
|
||||
extends HyperspaceIndexCRUDEvent
|
||||
extends HyperspaceIndexCRUDEvent
|
||||
|
||||
/**
|
||||
* Index Refresh Event for quick mode. Emitted when refresh is called on an index
|
||||
|
|
@ -125,7 +125,7 @@ case class RefreshIncrementalActionEvent(appInfo: AppInfo, index: IndexLogEntry,
|
|||
* @param message Message about event.
|
||||
*/
|
||||
case class RefreshQuickActionEvent(appInfo: AppInfo, index: IndexLogEntry, message: String)
|
||||
extends HyperspaceIndexCRUDEvent
|
||||
extends HyperspaceIndexCRUDEvent
|
||||
|
||||
/**
|
||||
* Index Optimize Event for index files.
|
||||
|
|
@ -135,7 +135,7 @@ case class RefreshQuickActionEvent(appInfo: AppInfo, index: IndexLogEntry, messa
|
|||
* @param message Message about event.
|
||||
*/
|
||||
case class OptimizeActionEvent(appInfo: AppInfo, index: IndexLogEntry, message: String)
|
||||
extends HyperspaceIndexCRUDEvent
|
||||
extends HyperspaceIndexCRUDEvent
|
||||
|
||||
/**
|
||||
* Index usage event. This event is emitted when an index is picked instead of original data
|
||||
|
|
|
|||
|
|
@ -94,7 +94,9 @@ object ResolverUtils {
|
|||
*/
|
||||
def apply(normalizedColumnName: String): ResolvedColumn = {
|
||||
if (normalizedColumnName.startsWith(NESTED_FIELD_PREFIX)) {
|
||||
ResolvedColumn(normalizedColumnName.substring(NESTED_FIELD_PREFIX.length), isNested = true)
|
||||
ResolvedColumn(
|
||||
normalizedColumnName.substring(NESTED_FIELD_PREFIX.length),
|
||||
isNested = true)
|
||||
} else {
|
||||
ResolvedColumn(normalizedColumnName, isNested = false)
|
||||
}
|
||||
|
|
@ -212,20 +214,21 @@ object ResolverUtils {
|
|||
private def getColumnNameFromSchema(
|
||||
schema: StructType,
|
||||
resolvedColNameParts: Seq[String],
|
||||
resolver: Resolver): Seq[String] = resolvedColNameParts match {
|
||||
case h :: tail =>
|
||||
val field = schema.find(f => resolver(f.name, h)).get
|
||||
field match {
|
||||
case StructField(name, s: StructType, _, _) =>
|
||||
name +: getColumnNameFromSchema(s, tail, resolver)
|
||||
case StructField(_, _: ArrayType, _, _) =>
|
||||
// TODO: Nested arrays will be supported later
|
||||
throw HyperspaceException("Array types are not supported.")
|
||||
case StructField(_, _: MapType, _, _) =>
|
||||
// TODO: Nested maps will be supported later
|
||||
throw HyperspaceException("Map types are not supported")
|
||||
case f => Seq(f.name)
|
||||
}
|
||||
case _ => Nil
|
||||
}
|
||||
resolver: Resolver): Seq[String] =
|
||||
resolvedColNameParts match {
|
||||
case h :: tail =>
|
||||
val field = schema.find(f => resolver(f.name, h)).get
|
||||
field match {
|
||||
case StructField(name, s: StructType, _, _) =>
|
||||
name +: getColumnNameFromSchema(s, tail, resolver)
|
||||
case StructField(_, _: ArrayType, _, _) =>
|
||||
// TODO: Nested arrays will be supported later
|
||||
throw HyperspaceException("Array types are not supported.")
|
||||
case StructField(_, _: MapType, _, _) =>
|
||||
// TODO: Nested maps will be supported later
|
||||
throw HyperspaceException("Map types are not supported")
|
||||
case f => Seq(f.name)
|
||||
}
|
||||
case _ => Nil
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -24,35 +24,83 @@ import org.apache.spark.sql.SparkSession
|
|||
object SampleNestedData {
|
||||
|
||||
val testData = Seq(
|
||||
("2017-09-03", "810a20a2baa24ff3ad493bfbf064569a", "donde", 2, 1000,
|
||||
SampleNestedDataStruct("id1", SampleNestedDataLeaf("leaf_id1", 1))),
|
||||
("2017-09-03", "fd093f8a05604515957083e70cb3dceb", "facebook", 1, 3000,
|
||||
SampleNestedDataStruct("id1", SampleNestedDataLeaf("leaf_id1", 2))),
|
||||
("2017-09-03", "af3ed6a197a8447cba8bc8ea21fad208", "facebook", 1, 3000,
|
||||
SampleNestedDataStruct("id2", SampleNestedDataLeaf("leaf_id7", 1))),
|
||||
("2017-09-03", "975134eca06c4711a0406d0464cbe7d6", "facebook", 1, 4000,
|
||||
SampleNestedDataStruct("id2", SampleNestedDataLeaf("leaf_id7", 2))),
|
||||
("2018-09-03", "e90a6028e15b4f4593eef557daf5166d", "ibraco", 2, 3000,
|
||||
SampleNestedDataStruct("id2", SampleNestedDataLeaf("leaf_id7", 5))),
|
||||
("2018-09-03", "576ed96b0d5340aa98a47de15c9f87ce", "facebook", 2, 3000,
|
||||
SampleNestedDataStruct("id2", SampleNestedDataLeaf("leaf_id9", 1))),
|
||||
("2018-09-03", "50d690516ca641438166049a6303650c", "ibraco", 2, 1000,
|
||||
SampleNestedDataStruct("id3", SampleNestedDataLeaf("leaf_id9", 10))),
|
||||
("2019-10-03", "380786e6495d4cd8a5dd4cc8d3d12917", "facebook", 2, 3000,
|
||||
SampleNestedDataStruct("id4", SampleNestedDataLeaf("leaf_id9", 12))),
|
||||
("2019-10-03", "ff60e4838b92421eafc3e6ee59a9e9f1", "miperro", 2, 2000,
|
||||
SampleNestedDataStruct("id5", SampleNestedDataLeaf("leaf_id9", 21))),
|
||||
("2019-10-03", "187696fe0a6a40cc9516bc6e47c70bc1", "facebook", 4, 3000,
|
||||
SampleNestedDataStruct("id6", SampleNestedDataLeaf("leaf_id9", 22))))
|
||||
(
|
||||
"2017-09-03",
|
||||
"810a20a2baa24ff3ad493bfbf064569a",
|
||||
"donde",
|
||||
2,
|
||||
1000,
|
||||
SampleNestedDataStruct("id1", SampleNestedDataLeaf("leaf_id1", 1))),
|
||||
(
|
||||
"2017-09-03",
|
||||
"fd093f8a05604515957083e70cb3dceb",
|
||||
"facebook",
|
||||
1,
|
||||
3000,
|
||||
SampleNestedDataStruct("id1", SampleNestedDataLeaf("leaf_id1", 2))),
|
||||
(
|
||||
"2017-09-03",
|
||||
"af3ed6a197a8447cba8bc8ea21fad208",
|
||||
"facebook",
|
||||
1,
|
||||
3000,
|
||||
SampleNestedDataStruct("id2", SampleNestedDataLeaf("leaf_id7", 1))),
|
||||
(
|
||||
"2017-09-03",
|
||||
"975134eca06c4711a0406d0464cbe7d6",
|
||||
"facebook",
|
||||
1,
|
||||
4000,
|
||||
SampleNestedDataStruct("id2", SampleNestedDataLeaf("leaf_id7", 2))),
|
||||
(
|
||||
"2018-09-03",
|
||||
"e90a6028e15b4f4593eef557daf5166d",
|
||||
"ibraco",
|
||||
2,
|
||||
3000,
|
||||
SampleNestedDataStruct("id2", SampleNestedDataLeaf("leaf_id7", 5))),
|
||||
(
|
||||
"2018-09-03",
|
||||
"576ed96b0d5340aa98a47de15c9f87ce",
|
||||
"facebook",
|
||||
2,
|
||||
3000,
|
||||
SampleNestedDataStruct("id2", SampleNestedDataLeaf("leaf_id9", 1))),
|
||||
(
|
||||
"2018-09-03",
|
||||
"50d690516ca641438166049a6303650c",
|
||||
"ibraco",
|
||||
2,
|
||||
1000,
|
||||
SampleNestedDataStruct("id3", SampleNestedDataLeaf("leaf_id9", 10))),
|
||||
(
|
||||
"2019-10-03",
|
||||
"380786e6495d4cd8a5dd4cc8d3d12917",
|
||||
"facebook",
|
||||
2,
|
||||
3000,
|
||||
SampleNestedDataStruct("id4", SampleNestedDataLeaf("leaf_id9", 12))),
|
||||
(
|
||||
"2019-10-03",
|
||||
"ff60e4838b92421eafc3e6ee59a9e9f1",
|
||||
"miperro",
|
||||
2,
|
||||
2000,
|
||||
SampleNestedDataStruct("id5", SampleNestedDataLeaf("leaf_id9", 21))),
|
||||
(
|
||||
"2019-10-03",
|
||||
"187696fe0a6a40cc9516bc6e47c70bc1",
|
||||
"facebook",
|
||||
4,
|
||||
3000,
|
||||
SampleNestedDataStruct("id6", SampleNestedDataLeaf("leaf_id9", 22))))
|
||||
|
||||
def save(
|
||||
spark: SparkSession,
|
||||
path: String,
|
||||
columns: Seq[String],
|
||||
partitionColumns: Option[Seq[String]] = None): Unit = {
|
||||
val df = spark.createDataFrame(
|
||||
spark.sparkContext.parallelize(testData)
|
||||
).toDF(columns: _*)
|
||||
val df = spark.createDataFrame(spark.sparkContext.parallelize(testData)).toDF(columns: _*)
|
||||
partitionColumns match {
|
||||
case Some(pcs) =>
|
||||
df.write.partitionBy(pcs: _*).parquet(path)
|
||||
|
|
|
|||
|
|
@ -58,7 +58,6 @@ import com.microsoft.hyperspace.util.SparkTestShims.SimpleExplainCommand
|
|||
* The explain files are saved to help debug later, they are not checked. Only the simplified
|
||||
* plans are checked (by string comparison).
|
||||
*
|
||||
*
|
||||
* To run the entire test suite:
|
||||
* {{{
|
||||
* sbt "test:testOnly *PlanStabilitySuite"
|
||||
|
|
@ -178,14 +177,15 @@ trait PlanStabilitySuite extends TPCDSBase with SQLHelper with Logging {
|
|||
val exchangeIdMap = new mutable.HashMap[SparkPlan, Int]()
|
||||
val subqueriesMap = new mutable.HashMap[SparkPlan, Int]()
|
||||
|
||||
def getId(plan: SparkPlan): Int = plan match {
|
||||
case exchange: Exchange => exchangeIdMap.getOrElseUpdate(exchange, exchangeIdMap.size + 1)
|
||||
case ReusedExchangeExec(_, exchange) =>
|
||||
exchangeIdMap.getOrElseUpdate(exchange, exchangeIdMap.size + 1)
|
||||
case subquery: SubqueryExec =>
|
||||
subqueriesMap.getOrElseUpdate(subquery, subqueriesMap.size + 1)
|
||||
case _ => -1
|
||||
}
|
||||
def getId(plan: SparkPlan): Int =
|
||||
plan match {
|
||||
case exchange: Exchange => exchangeIdMap.getOrElseUpdate(exchange, exchangeIdMap.size + 1)
|
||||
case ReusedExchangeExec(_, exchange) =>
|
||||
exchangeIdMap.getOrElseUpdate(exchange, exchangeIdMap.size + 1)
|
||||
case subquery: SubqueryExec =>
|
||||
subqueriesMap.getOrElseUpdate(subquery, subqueriesMap.size + 1)
|
||||
case _ => -1
|
||||
}
|
||||
|
||||
/**
|
||||
* Some expression names have ExprId in them due to using things such as
|
||||
|
|
@ -273,7 +273,9 @@ trait PlanStabilitySuite extends TPCDSBase with SQLHelper with Logging {
|
|||
*/
|
||||
class TPCDSV1_4_SparkPlanStabilitySuite extends PlanStabilitySuite {
|
||||
override val goldenFilePath: String = {
|
||||
new File(baseResourcePath, s"spark-${BuildInfo.sparkShortVersion}/approved-plans-v1_4").getAbsolutePath
|
||||
new File(
|
||||
baseResourcePath,
|
||||
s"spark-${BuildInfo.sparkShortVersion}/approved-plans-v1_4").getAbsolutePath
|
||||
}
|
||||
|
||||
// Enable cross join because some queries fail during query optimization phase.
|
||||
|
|
|
|||
|
|
@ -536,8 +536,7 @@ trait TPCDSBase extends SparkFunSuite with SparkInvolvedSuite {
|
|||
|`t_shift` STRING,
|
||||
|`t_sub_shift` STRING,
|
||||
|`t_meal_time` STRING
|
||||
""".stripMargin
|
||||
)
|
||||
""".stripMargin)
|
||||
|
||||
val tableNames: Iterable[String] = tableColumns.keys
|
||||
|
||||
|
|
@ -546,8 +545,7 @@ trait TPCDSBase extends SparkFunSuite with SparkInvolvedSuite {
|
|||
tableName: String,
|
||||
format: String = "parquet",
|
||||
options: Seq[String] = Nil): Unit = {
|
||||
spark.sql(
|
||||
s"""
|
||||
spark.sql(s"""
|
||||
|CREATE TABLE `$tableName` (${tableColumns(tableName)})
|
||||
|USING $format
|
||||
|${options.mkString("\n")}
|
||||
|
|
|
|||
|
|
@ -109,7 +109,7 @@ class BucketUnionTest extends SparkFunSuite with SparkInvolvedSuite {
|
|||
val rdd = new BucketUnionRDD[Row](spark.sparkContext, Seq(p1.rdd, p2.rdd), bucketSpec)
|
||||
assert(
|
||||
rdd.collect.sortBy(r => (r.getInt(0), r.getString(1))).map(r => r.toSeq.toList).toList
|
||||
== Seq(Seq(2, "name1"), Seq(2, "name3"), Seq(3, "name2"), Seq(3, "name4")))
|
||||
== Seq(Seq(2, "name1"), Seq(2, "name3"), Seq(3, "name2"), Seq(3, "name4")))
|
||||
assert(rdd.getPartitions.length == 10)
|
||||
assert(rdd.partitions.head.isInstanceOf[BucketUnionRDDPartition])
|
||||
|
||||
|
|
|
|||
|
|
@ -141,10 +141,7 @@ class CreateIndexNestedTest extends HyperspaceSuite with SQLHelper {
|
|||
val dfB = nonPartitionedDataDF.as("B")
|
||||
val dfJoin = dfA
|
||||
.join(dfB, dfA("Query") === dfB("Query"))
|
||||
.select(
|
||||
dfA("RGUID"),
|
||||
dfA("Query"),
|
||||
dfA("nested.leaf.cnt"))
|
||||
.select(dfA("RGUID"), dfA("Query"), dfA("nested.leaf.cnt"))
|
||||
val exception = intercept[HyperspaceException] {
|
||||
hyperspace.createIndex(dfJoin, indexConfig1)
|
||||
}
|
||||
|
|
@ -153,7 +150,8 @@ class CreateIndexNestedTest extends HyperspaceSuite with SQLHelper {
|
|||
"Only creating index over HDFS file based scan nodes is supported."))
|
||||
}
|
||||
|
||||
test("Check lineage in index records for partitioned data when partition key is not in config.") {
|
||||
test(
|
||||
"Check lineage in index records for partitioned data when partition key is not in config.") {
|
||||
withSQLConf(IndexConstants.INDEX_LINEAGE_ENABLED -> "true") {
|
||||
hyperspace.createIndex(partitionedDataDF, indexConfig2)
|
||||
val indexRecordsDF = spark.read.parquet(
|
||||
|
|
|
|||
|
|
@ -145,10 +145,7 @@ class CreateIndexTest extends HyperspaceSuite with SQLHelper {
|
|||
val dfB = nonPartitionedDataDF.as("B")
|
||||
val dfJoin = dfA
|
||||
.join(dfB, dfA("Query") === dfB("Query"))
|
||||
.select(
|
||||
dfA("RGUID"),
|
||||
dfA("Query"),
|
||||
dfA("imprs"))
|
||||
.select(dfA("RGUID"), dfA("Query"), dfA("imprs"))
|
||||
val exception = intercept[HyperspaceException] {
|
||||
hyperspace.createIndex(dfJoin, indexConfig1)
|
||||
}
|
||||
|
|
@ -171,7 +168,8 @@ class CreateIndexTest extends HyperspaceSuite with SQLHelper {
|
|||
}
|
||||
}
|
||||
|
||||
test("Check lineage in index records for partitioned data when partition key is not in config.") {
|
||||
test(
|
||||
"Check lineage in index records for partitioned data when partition key is not in config.") {
|
||||
withSQLConf(IndexConstants.INDEX_LINEAGE_ENABLED -> "true") {
|
||||
hyperspace.createIndex(partitionedDataDF, indexConfig3)
|
||||
val indexRecordsDF = spark.read.parquet(
|
||||
|
|
@ -201,7 +199,8 @@ class CreateIndexTest extends HyperspaceSuite with SQLHelper {
|
|||
}
|
||||
}
|
||||
|
||||
test("Check lineage in index records for partitioned data when partition key is in load path.") {
|
||||
test(
|
||||
"Check lineage in index records for partitioned data when partition key is in load path.") {
|
||||
withSQLConf(IndexConstants.INDEX_LINEAGE_ENABLED -> "true") {
|
||||
val dataDF =
|
||||
spark.read.parquet(s"$partitionedDataPath/${partitionKeys.head}=2017-09-03")
|
||||
|
|
|
|||
|
|
@ -549,10 +549,10 @@ class DeltaLakeIntegrationTest extends QueryTest with HyperspaceSuite {
|
|||
private def getDeltaLakeTableVersion(df: DataFrame): Long = {
|
||||
df.queryExecution.optimizedPlan match {
|
||||
case LogicalRelation(
|
||||
HadoopFsRelation(location: TahoeLogFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
HadoopFsRelation(location: TahoeLogFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
location.tableVersion
|
||||
}
|
||||
}
|
||||
|
|
@ -586,10 +586,10 @@ class DeltaLakeIntegrationTest extends QueryTest with HyperspaceSuite {
|
|||
def isIndexUsed(plan: LogicalPlan, expectedPathsSubStr: String*): Boolean = {
|
||||
val rootPaths = plan.collect {
|
||||
case LogicalRelation(
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
location.rootPaths
|
||||
}.flatten
|
||||
rootPaths.nonEmpty && rootPaths.forall(p =>
|
||||
|
|
|
|||
|
|
@ -435,7 +435,8 @@ class E2EHyperspaceRulesTest extends QueryTest with HyperspaceSuite {
|
|||
sortedRowsWithHyperspaceDisabled.sameElements(getSortedRows(dfAfterHyperspaceDisabled)))
|
||||
}
|
||||
|
||||
test("Verify JoinIndexRule utilizes indexes correctly after incremental refresh (append-only).") {
|
||||
test(
|
||||
"Verify JoinIndexRule utilizes indexes correctly after incremental refresh (append-only).") {
|
||||
withTempPathAsString { testPath =>
|
||||
// Setup. Create data.
|
||||
val indexConfig = IndexConfig("index", Seq("c2"), Seq("c4"))
|
||||
|
|
@ -596,7 +597,8 @@ class E2EHyperspaceRulesTest extends QueryTest with HyperspaceSuite {
|
|||
verifyIndexUsage(
|
||||
query,
|
||||
getIndexFilesPath(indexConfig.indexName, Seq(1)) ++ // for Left
|
||||
getIndexFilesPath(indexConfig.indexName, Seq(1))) // for Right
|
||||
getIndexFilesPath(indexConfig.indexName, Seq(1))
|
||||
) // for Right
|
||||
|
||||
// Verify correctness of results.
|
||||
spark.disableHyperspace()
|
||||
|
|
@ -659,7 +661,8 @@ class E2EHyperspaceRulesTest extends QueryTest with HyperspaceSuite {
|
|||
verifyIndexUsage(
|
||||
query,
|
||||
getIndexFilesPath(indexConfig.indexName, Seq(0)) ++ appendedFiles ++
|
||||
getIndexFilesPath(indexConfig.indexName, Seq(0)) ++ appendedFiles) // for Right
|
||||
getIndexFilesPath(indexConfig.indexName, Seq(0)) ++ appendedFiles
|
||||
) // for Right
|
||||
|
||||
// Verify correctness of results.
|
||||
spark.disableHyperspace()
|
||||
|
|
@ -1021,10 +1024,10 @@ class E2EHyperspaceRulesTest extends QueryTest with HyperspaceSuite {
|
|||
private def getAllRootPaths(optimizedPlan: LogicalPlan): Seq[Path] = {
|
||||
optimizedPlan.collect {
|
||||
case LogicalRelation(
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
location.rootPaths
|
||||
}.flatten
|
||||
}
|
||||
|
|
|
|||
|
|
@ -58,14 +58,15 @@ class FileIdTrackerTest extends SparkFunSuite {
|
|||
val tracker = new FileIdTracker
|
||||
tracker.addFileInfo(Set(FileInfo("def", 123, 555, 10)))
|
||||
val ex = intercept[HyperspaceException] {
|
||||
implicit def ordering: Ordering[FileInfo] = new Ordering[FileInfo] {
|
||||
override def compare(x: FileInfo, y: FileInfo): Int = {
|
||||
x.name.compareTo(y.name)
|
||||
implicit def ordering: Ordering[FileInfo] =
|
||||
new Ordering[FileInfo] {
|
||||
override def compare(x: FileInfo, y: FileInfo): Int = {
|
||||
x.name.compareTo(y.name)
|
||||
}
|
||||
}
|
||||
}
|
||||
tracker.addFileInfo(scala.collection.immutable.SortedSet(
|
||||
FileInfo("abc", 100, 555, 15),
|
||||
FileInfo("def", 123, 555, 11)))
|
||||
tracker.addFileInfo(
|
||||
scala.collection.immutable
|
||||
.SortedSet(FileInfo("abc", 100, 555, 15), FileInfo("def", 123, 555, 11)))
|
||||
}
|
||||
assert(ex.getMessage.contains("Adding file info with a conflicting id"))
|
||||
assert(tracker.getFileId("abc", 100, 555).contains(15))
|
||||
|
|
|
|||
|
|
@ -142,14 +142,16 @@ class HybridScanForNonPartitionedDataTest extends HybridScanSuite {
|
|||
val deletedRatio = 1 - (afterDeleteSize / sourceSize.toFloat)
|
||||
|
||||
withSQLConf(TestConfig.HybridScanEnabled: _*) {
|
||||
withSQLConf(IndexConstants.INDEX_HYBRID_SCAN_DELETED_RATIO_THRESHOLD ->
|
||||
(deletedRatio + 0.1).toString) {
|
||||
withSQLConf(
|
||||
IndexConstants.INDEX_HYBRID_SCAN_DELETED_RATIO_THRESHOLD ->
|
||||
(deletedRatio + 0.1).toString) {
|
||||
val filter = filterQuery
|
||||
// As deletedRatio is less than the threshold, the index can be applied.
|
||||
assert(!basePlan.equals(filter.queryExecution.optimizedPlan))
|
||||
}
|
||||
withSQLConf(IndexConstants.INDEX_HYBRID_SCAN_DELETED_RATIO_THRESHOLD ->
|
||||
(deletedRatio - 0.1).toString) {
|
||||
withSQLConf(
|
||||
IndexConstants.INDEX_HYBRID_SCAN_DELETED_RATIO_THRESHOLD ->
|
||||
(deletedRatio - 0.1).toString) {
|
||||
val filter = filterQuery
|
||||
// As deletedRatio is greater than the threshold, the index shouldn't be applied.
|
||||
assert(basePlan.equals(filter.queryExecution.optimizedPlan))
|
||||
|
|
|
|||
|
|
@ -129,8 +129,8 @@ trait HybridScanSuite extends QueryTest with HyperspaceSuite {
|
|||
}.flatten
|
||||
val deletedFilesList = plan collect {
|
||||
case Filter(
|
||||
Not(EqualTo(left: Attribute, right: Literal)),
|
||||
LogicalRelation(fsRelation: HadoopFsRelation, _, _, _)) =>
|
||||
Not(EqualTo(left: Attribute, right: Literal)),
|
||||
LogicalRelation(fsRelation: HadoopFsRelation, _, _, _)) =>
|
||||
// Check new filter condition on lineage column.
|
||||
val colName = left.toString
|
||||
val deletedFile = right.toString
|
||||
|
|
@ -142,8 +142,8 @@ trait HybridScanSuite extends QueryTest with HyperspaceSuite {
|
|||
assert(files.nonEmpty && files.forall(_.contains(indexName)))
|
||||
deleted
|
||||
case Filter(
|
||||
Not(InSet(attr, deletedFileIds)),
|
||||
LogicalRelation(fsRelation: HadoopFsRelation, _, _, _)) =>
|
||||
Not(InSet(attr, deletedFileIds)),
|
||||
LogicalRelation(fsRelation: HadoopFsRelation, _, _, _)) =>
|
||||
// Check new filter condition on lineage column.
|
||||
assert(attr.toString.contains(IndexConstants.DATA_FILE_NAME_ID))
|
||||
val deleted = deletedFileIds.map(_.toString).toSeq
|
||||
|
|
@ -156,8 +156,8 @@ trait HybridScanSuite extends QueryTest with HyperspaceSuite {
|
|||
assert(files.nonEmpty && files.forall(_.contains(indexName)))
|
||||
deleted
|
||||
case Filter(
|
||||
Not(In(attr, deletedFileIds)),
|
||||
LogicalRelation(fsRelation: HadoopFsRelation, _, _, _)) =>
|
||||
Not(In(attr, deletedFileIds)),
|
||||
LogicalRelation(fsRelation: HadoopFsRelation, _, _, _)) =>
|
||||
// Check new filter condition on lineage column.
|
||||
assert(attr.toString.contains(IndexConstants.DATA_FILE_NAME_ID))
|
||||
val deleted = deletedFileIds.map(_.toString)
|
||||
|
|
@ -290,7 +290,8 @@ trait HybridScanSuite extends QueryTest with HyperspaceSuite {
|
|||
case p @ BucketUnionExec(children, bucketSpec) =>
|
||||
assert(children.size === 2)
|
||||
// children.head is always the index plan.
|
||||
assert(children.head.isInstanceOf[ProjectExec] || children.head.isInstanceOf[FilterExec])
|
||||
assert(
|
||||
children.head.isInstanceOf[ProjectExec] || children.head.isInstanceOf[FilterExec])
|
||||
assert(children.last.isInstanceOf[ShuffleExchangeExec])
|
||||
assert(bucketSpec.numBuckets === 200)
|
||||
p
|
||||
|
|
@ -582,7 +583,8 @@ trait HybridScanSuite extends QueryTest with HyperspaceSuite {
|
|||
}
|
||||
}
|
||||
|
||||
test("Delete-only: join rule, deleted files should be excluded from each index data relation.") {
|
||||
test(
|
||||
"Delete-only: join rule, deleted files should be excluded from each index data relation.") {
|
||||
withTempPathAsString { testPath =>
|
||||
val deletePath1 = testPath + "/delete1"
|
||||
val deletePath2 = testPath + "/delete2"
|
||||
|
|
|
|||
|
|
@ -376,10 +376,10 @@ class IcebergIntegrationTest extends QueryTest with HyperspaceSuite {
|
|||
def isIndexUsed(plan: LogicalPlan, expectedPathsSubStr: String*): Boolean = {
|
||||
val rootPaths = plan.collect {
|
||||
case LogicalRelation(
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
location.rootPaths
|
||||
}.flatten
|
||||
rootPaths.nonEmpty && rootPaths.forall(p =>
|
||||
|
|
|
|||
|
|
@ -48,12 +48,7 @@ class IndexCacheTest extends HyperspaceSuite {
|
|||
|
||||
val entry = IndexLogEntry(
|
||||
"index1",
|
||||
CoveringIndex(
|
||||
Seq("RGUID"),
|
||||
Seq("Date"),
|
||||
schema,
|
||||
10,
|
||||
Map()),
|
||||
CoveringIndex(Seq("RGUID"), Seq("Date"), schema, 10, Map()),
|
||||
Content(Directory(indexDir)),
|
||||
Source(SparkPlan(sourcePlanProperties)),
|
||||
Map())
|
||||
|
|
@ -152,7 +147,7 @@ class IndexCacheTest extends HyperspaceSuite {
|
|||
|
||||
/**
|
||||
* Mock for testing purposes so we can validate and invalidate entries based on time.
|
||||
|
||||
*
|
||||
* @param time Current time.
|
||||
*/
|
||||
class MockClock(private var time: Long = 0L) extends Clock {
|
||||
|
|
|
|||
|
|
@ -47,12 +47,7 @@ class IndexCollectionManagerTest extends HyperspaceSuite {
|
|||
|
||||
val entry = IndexLogEntry(
|
||||
indexPath.toString,
|
||||
CoveringIndex(
|
||||
Seq("RGUID"),
|
||||
Seq("Date"),
|
||||
new StructType(),
|
||||
10,
|
||||
Map()),
|
||||
CoveringIndex(Seq("RGUID"), Seq("Date"), new StructType(), 10, Map()),
|
||||
Content(Directory(s"$indexPath/${IndexConstants.INDEX_VERSION_DIRECTORY_PREFIX}=0")),
|
||||
Source(SparkPlan(sourcePlanProperties)),
|
||||
Map())
|
||||
|
|
@ -98,12 +93,7 @@ class IndexCollectionManagerTest extends HyperspaceSuite {
|
|||
|
||||
val entry = IndexLogEntry(
|
||||
str,
|
||||
CoveringIndex(
|
||||
Seq("RGUID"),
|
||||
Seq("Date"),
|
||||
new StructType(),
|
||||
10,
|
||||
Map()),
|
||||
CoveringIndex(Seq("RGUID"), Seq("Date"), new StructType(), 10, Map()),
|
||||
Content(Directory(s"$str/${IndexConstants.INDEX_VERSION_DIRECTORY_PREFIX}=0")),
|
||||
Source(SparkPlan(sourcePlanProperties)),
|
||||
Map())
|
||||
|
|
|
|||
|
|
@ -193,43 +193,24 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
Seq(
|
||||
Relation(
|
||||
Seq("rootpath"),
|
||||
Hdfs(
|
||||
Hdfs.Properties(
|
||||
Content(
|
||||
Directory(
|
||||
"test",
|
||||
Seq(FileInfo("f1", 100L, 100L, 0), FileInfo("f2", 100L, 200L, 1)),
|
||||
Seq()
|
||||
)
|
||||
),
|
||||
Some(
|
||||
Update(
|
||||
None,
|
||||
Some(Content(Directory("", Seq(FileInfo("f1", 10, 10, 2)))))
|
||||
)
|
||||
)
|
||||
)
|
||||
),
|
||||
Hdfs(Hdfs.Properties(
|
||||
Content(Directory(
|
||||
"test",
|
||||
Seq(FileInfo("f1", 100L, 100L, 0), FileInfo("f2", 100L, 200L, 1)),
|
||||
Seq())),
|
||||
Some(Update(None, Some(Content(Directory("", Seq(FileInfo("f1", 10, 10, 2))))))))),
|
||||
new StructType(),
|
||||
"type",
|
||||
Map()
|
||||
)
|
||||
),
|
||||
Map())),
|
||||
null,
|
||||
null,
|
||||
LogicalPlanFingerprint(
|
||||
LogicalPlanFingerprint
|
||||
.Properties(Seq(Signature("provider", "signatureValue")))
|
||||
))
|
||||
.Properties(Seq(Signature("provider", "signatureValue")))))
|
||||
|
||||
val expected = IndexLogEntry.create(
|
||||
"indexName",
|
||||
CoveringIndex(
|
||||
Seq("col1"),
|
||||
Seq("col2", "col3"),
|
||||
schema,
|
||||
200,
|
||||
Map()),
|
||||
CoveringIndex(Seq("col1"), Seq("col2", "col3"), schema, 200, Map()),
|
||||
Content(Directory("rootContentPath")),
|
||||
Source(SparkPlan(expectedSourcePlanProperties)),
|
||||
Map())
|
||||
|
|
@ -242,15 +223,19 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
}
|
||||
|
||||
test("Content.files api lists all files from Content object.") {
|
||||
val content = Content(Directory("file:/", subDirs =
|
||||
Seq(
|
||||
Directory("a",
|
||||
files = Seq(FileInfo("f1", 0, 0, UNKNOWN_FILE_ID), FileInfo("f2", 0, 0, UNKNOWN_FILE_ID)),
|
||||
val content = Content(
|
||||
Directory(
|
||||
"file:/",
|
||||
subDirs = Seq(Directory(
|
||||
"a",
|
||||
files =
|
||||
Seq(FileInfo("f1", 0, 0, UNKNOWN_FILE_ID), FileInfo("f2", 0, 0, UNKNOWN_FILE_ID)),
|
||||
subDirs = Seq(
|
||||
Directory("b",
|
||||
files =
|
||||
Seq(FileInfo("f3", 0, 0, UNKNOWN_FILE_ID), FileInfo("f4", 0, 0, UNKNOWN_FILE_ID)))))
|
||||
)))
|
||||
Directory(
|
||||
"b",
|
||||
files = Seq(
|
||||
FileInfo("f3", 0, 0, UNKNOWN_FILE_ID),
|
||||
FileInfo("f4", 0, 0, UNKNOWN_FILE_ID))))))))
|
||||
|
||||
val expected =
|
||||
Seq("file:/a/f1", "file:/a/f2", "file:/a/b/f3", "file:/a/b/f4").map(new Path(_)).toSet
|
||||
|
|
@ -262,8 +247,9 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
val nestedDirPath = toPath(nestedDir)
|
||||
|
||||
val expected = {
|
||||
val fileInfos = Seq(f3, f4).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val fileInfos = Seq(f3, f4)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val nestedDirDirectory = Directory("nested", fileInfos)
|
||||
val rootDirectory = createDirectory(nestedDirPath, nestedDirDirectory)
|
||||
Content(rootDirectory, NoOpFingerprint())
|
||||
|
|
@ -277,8 +263,9 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
val nestedDirPath = toPath(nestedDir)
|
||||
|
||||
val expected = {
|
||||
val fileInfos = Seq(f3, f4).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val fileInfos = Seq(f3, f4)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val nestedDirDirectory = Directory("nested", fileInfos)
|
||||
val rootDirectory = createDirectory(nestedDirPath, nestedDirDirectory)
|
||||
Content(rootDirectory, NoOpFingerprint())
|
||||
|
|
@ -292,8 +279,9 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
val nestedDirPath = toPath(nestedDir)
|
||||
|
||||
val expected = {
|
||||
val fileInfos = Seq(f3, f4).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val fileInfos = Seq(f3, f4)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val nestedDirDirectory = Directory("nested", fileInfos)
|
||||
createDirectory(nestedDirPath, nestedDirDirectory)
|
||||
}
|
||||
|
|
@ -302,17 +290,21 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
assert(directoryEquals(actual, expected))
|
||||
}
|
||||
|
||||
test("Directory.fromDirectory api creates the correct Directory objects, " +
|
||||
"recursively listing all leaf files.") {
|
||||
test(
|
||||
"Directory.fromDirectory api creates the correct Directory objects, " +
|
||||
"recursively listing all leaf files.") {
|
||||
val testDirPath = toPath(testDir)
|
||||
|
||||
val testDirLeafFiles =
|
||||
Seq(f1, f2).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
Seq(f1, f2)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val nestedDirLeafFiles =
|
||||
Seq(f3, f4).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val testDirDirectory = Directory(name = "testDir",
|
||||
Seq(f3, f4)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val testDirDirectory = Directory(
|
||||
name = "testDir",
|
||||
files = testDirLeafFiles,
|
||||
subDirs = Seq(Directory(name = "nested", files = nestedDirLeafFiles)))
|
||||
val expected = createDirectory(testDirPath, testDirDirectory)
|
||||
|
|
@ -326,12 +318,15 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
val testDirPath = toPath(testDir)
|
||||
|
||||
val testDirLeafFiles =
|
||||
Seq(f1, f2).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
Seq(f1, f2)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val nestedDirLeafFiles =
|
||||
Seq(f3, f4).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val testDirDirectory = Directory(name = "testDir",
|
||||
Seq(f3, f4)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val testDirDirectory = Directory(
|
||||
name = "testDir",
|
||||
files = testDirLeafFiles,
|
||||
subDirs = Seq(Directory(name = "nested", files = nestedDirLeafFiles)))
|
||||
|
||||
|
|
@ -345,12 +340,15 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
test("Directory.fromLeafFiles api does not include other files in the directory.") {
|
||||
val testDirPath = toPath(testDir)
|
||||
|
||||
val testDirLeafFiles = Seq(f1).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val testDirLeafFiles = Seq(f1)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val nestedDirLeafFiles =
|
||||
Seq(f4).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val testDirDirectory = Directory(name = "testDir",
|
||||
Seq(f4)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val testDirDirectory = Directory(
|
||||
name = "testDir",
|
||||
files = testDirLeafFiles,
|
||||
subDirs = Seq(Directory(name = "nested", files = nestedDirLeafFiles)))
|
||||
|
||||
|
|
@ -389,8 +387,9 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
override def accept(path: Path): Boolean = path.getName.startsWith("f1")
|
||||
}
|
||||
|
||||
val testDirLeafFiles = Seq(f1).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val testDirLeafFiles = Seq(f1)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false))
|
||||
val testDirDirectory = Directory(name = "testDir", files = testDirLeafFiles)
|
||||
val expected = createDirectory(testDirPath, testDirDirectory)
|
||||
|
||||
|
|
@ -400,8 +399,9 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
assert(directoryEquals(actual, expected))
|
||||
}
|
||||
|
||||
test("Directory.fromDirectory and fromLeafFileswhere files are at same level but different" +
|
||||
"dirs.") {
|
||||
test(
|
||||
"Directory.fromDirectory and fromLeafFileswhere files are at same level but different" +
|
||||
"dirs.") {
|
||||
// File Structure
|
||||
// testDir/temp/a/f1
|
||||
// testDir/temp/b/f2
|
||||
|
|
@ -413,11 +413,17 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
val f2 = Files.createFile(Paths.get(b + "/f2"))
|
||||
|
||||
val aDirectory =
|
||||
Directory("a", Seq(f1).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
Directory(
|
||||
"a",
|
||||
Seq(f1)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
val bDirectory =
|
||||
Directory("b", Seq(f2).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
Directory(
|
||||
"b",
|
||||
Seq(f2)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
val tempDirectory = Directory("temp", subDirs = Seq(aDirectory, bDirectory))
|
||||
val tempDirectoryPath = toPath(tempDir)
|
||||
|
||||
|
|
@ -445,12 +451,18 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
val f2 = Files.createFile(Paths.get(c + "/f2"))
|
||||
|
||||
val cDirectory =
|
||||
Directory("c", Seq(f2).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
Directory(
|
||||
"c",
|
||||
Seq(f2)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
val bDirectory = Directory("b", subDirs = Seq(cDirectory))
|
||||
val aDirectory =
|
||||
Directory("a", Seq(f1).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
Directory(
|
||||
"a",
|
||||
Seq(f1)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
|
||||
val tempDirectory = Directory("temp", subDirs = Seq(aDirectory, bDirectory))
|
||||
val tempDirectoryPath = toPath(tempDir)
|
||||
|
|
@ -465,8 +477,9 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
FileUtils.deleteDirectory(tempDir.toFile)
|
||||
}
|
||||
|
||||
test("Directory.fromDirectory and fromLeafFiles where files belong to multiple" +
|
||||
"subdirectories.") {
|
||||
test(
|
||||
"Directory.fromDirectory and fromLeafFiles where files belong to multiple" +
|
||||
"subdirectories.") {
|
||||
// File Structure
|
||||
// testDir/temp/a/f1
|
||||
// testDir/temp/a/b/f2
|
||||
|
|
@ -481,17 +494,23 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
val f3 = Files.createFile(Paths.get(c + "/f3"))
|
||||
|
||||
val bDirectory =
|
||||
Directory("b", Seq(f2).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
Directory(
|
||||
"b",
|
||||
Seq(f2)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
val cDirectory =
|
||||
Directory("c", Seq(f3).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
Directory(
|
||||
"c",
|
||||
Seq(f3)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)))
|
||||
val aDirectory = Directory(
|
||||
"a",
|
||||
Seq(f1).map(toFileStatus).map(f =>
|
||||
FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)),
|
||||
Seq(bDirectory, cDirectory)
|
||||
)
|
||||
Seq(f1)
|
||||
.map(toFileStatus)
|
||||
.map(f => FileInfo(f, fileIdTracker.addFile(f), asFullPath = false)),
|
||||
Seq(bDirectory, cDirectory))
|
||||
val tempDirectory = Directory("temp", subDirs = Seq(aDirectory))
|
||||
val tempDirectoryPath = toPath(tempDir)
|
||||
|
||||
|
|
@ -511,11 +530,7 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
// a/f2
|
||||
val directory1 = Directory(
|
||||
name = "a",
|
||||
files = Seq(
|
||||
FileInfo("f1", 100L, 100L, 1L),
|
||||
FileInfo("f2", 100L, 100L, 2L)
|
||||
)
|
||||
)
|
||||
files = Seq(FileInfo("f1", 100L, 100L, 1L), FileInfo("f2", 100L, 100L, 2L)))
|
||||
|
||||
// directory2:
|
||||
// a/b/f3
|
||||
|
|
@ -525,13 +540,7 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
subDirs = Seq(
|
||||
Directory(
|
||||
name = "b",
|
||||
files = Seq(
|
||||
FileInfo("f3", 100L, 100L, 3L),
|
||||
FileInfo("f4", 100L, 100L, 4L)
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
files = Seq(FileInfo("f3", 100L, 100L, 3L), FileInfo("f4", 100L, 100L, 4L)))))
|
||||
|
||||
// Expected result of merging directory1 and directory2:
|
||||
// a/f1
|
||||
|
|
@ -540,20 +549,11 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
// a/b/f4
|
||||
val expected = Directory(
|
||||
name = "a",
|
||||
files = Seq(
|
||||
FileInfo("f1", 100L, 100L, 1L),
|
||||
FileInfo("f2", 100L, 100L, 2L)
|
||||
),
|
||||
files = Seq(FileInfo("f1", 100L, 100L, 1L), FileInfo("f2", 100L, 100L, 2L)),
|
||||
subDirs = Seq(
|
||||
Directory(
|
||||
name = "b",
|
||||
files = Seq(
|
||||
FileInfo("f3", 100L, 100L, 3L),
|
||||
FileInfo("f4", 100L, 100L, 4L)
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
files = Seq(FileInfo("f3", 100L, 100L, 3L), FileInfo("f4", 100L, 100L, 4L)))))
|
||||
|
||||
val actual1 = directory1.merge(directory2)
|
||||
val actual2 = directory2.merge(directory1)
|
||||
|
|
@ -569,14 +569,8 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
// a/b/f3
|
||||
val directory1 = Directory(
|
||||
name = "a",
|
||||
files = Seq(
|
||||
FileInfo("f1", 100L, 100L, 1L),
|
||||
FileInfo("f2", 100L, 100L, 2L)
|
||||
),
|
||||
subDirs = Seq(
|
||||
Directory(name = "b", files = Seq(FileInfo("f3", 100L, 100L, 3L)))
|
||||
)
|
||||
)
|
||||
files = Seq(FileInfo("f1", 100L, 100L, 1L), FileInfo("f2", 100L, 100L, 2L)),
|
||||
subDirs = Seq(Directory(name = "b", files = Seq(FileInfo("f3", 100L, 100L, 3L)))))
|
||||
|
||||
// directory2:
|
||||
// a/f4
|
||||
|
|
@ -589,17 +583,8 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
subDirs = Seq(
|
||||
Directory(
|
||||
name = "b",
|
||||
files = Seq(
|
||||
FileInfo("f5", 100L, 100L, 5L),
|
||||
FileInfo("f6", 100L, 100L, 6L)
|
||||
),
|
||||
subDirs = Seq(Directory(
|
||||
name = "c",
|
||||
files = Seq(FileInfo("f7", 100L, 100L, 7L))
|
||||
))
|
||||
)
|
||||
)
|
||||
)
|
||||
files = Seq(FileInfo("f5", 100L, 100L, 5L), FileInfo("f6", 100L, 100L, 6L)),
|
||||
subDirs = Seq(Directory(name = "c", files = Seq(FileInfo("f7", 100L, 100L, 7L)))))))
|
||||
|
||||
// Expected result of merging directory1 and directory2:
|
||||
// directory1:
|
||||
|
|
@ -615,23 +600,15 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
files = Seq(
|
||||
FileInfo("f1", 100L, 100L, 1L),
|
||||
FileInfo("f2", 100L, 100L, 2L),
|
||||
FileInfo("f4", 100L, 100L, 4L)
|
||||
),
|
||||
FileInfo("f4", 100L, 100L, 4L)),
|
||||
subDirs = Seq(
|
||||
Directory(
|
||||
name = "b",
|
||||
files = Seq(
|
||||
FileInfo("f3", 100L, 100L, 3L),
|
||||
FileInfo("f5", 100L, 100L, 5L),
|
||||
FileInfo("f6", 100L, 100L, 6L)
|
||||
),
|
||||
subDirs = Seq(
|
||||
Directory("c",
|
||||
files = Seq(FileInfo("f7", 100L, 100L, 7L)))
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
FileInfo("f6", 100L, 100L, 6L)),
|
||||
subDirs = Seq(Directory("c", files = Seq(FileInfo("f7", 100L, 100L, 7L)))))))
|
||||
|
||||
val actual1 = directory1.merge(directory2)
|
||||
val actual2 = directory2.merge(directory1)
|
||||
|
|
@ -646,19 +623,17 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
// a/f2
|
||||
val directory1 = Directory(
|
||||
name = "a",
|
||||
files = Seq(FileInfo("f1", 100L, 100L, 1L), FileInfo("f2", 100L, 100L, 2L))
|
||||
)
|
||||
files = Seq(FileInfo("f1", 100L, 100L, 1L), FileInfo("f2", 100L, 100L, 2L)))
|
||||
|
||||
// directory2:
|
||||
// b/f3
|
||||
// b/f4
|
||||
val directory2 = Directory(
|
||||
name = "b",
|
||||
files = Seq(FileInfo("f3", 100L, 100L, 3L), FileInfo("f4", 100L, 100L, 4L))
|
||||
)
|
||||
files = Seq(FileInfo("f3", 100L, 100L, 3L), FileInfo("f4", 100L, 100L, 4L)))
|
||||
|
||||
val ex1 = intercept[HyperspaceException] (directory1.merge(directory2))
|
||||
val ex2 = intercept[HyperspaceException] (directory2.merge(directory1))
|
||||
val ex1 = intercept[HyperspaceException](directory1.merge(directory2))
|
||||
val ex2 = intercept[HyperspaceException](directory2.merge(directory1))
|
||||
|
||||
assert(ex1.msg.contains("Merging directories with names a and b failed."))
|
||||
assert(ex2.msg.contains("Merging directories with names b and a failed."))
|
||||
|
|
@ -670,19 +645,18 @@ class IndexLogEntryTest extends HyperspaceSuite with SQLHelper {
|
|||
|
||||
private def directoryEquals(dir1: Directory, dir2: Directory): Boolean = {
|
||||
dir1.name.equals(dir2.name) &&
|
||||
dir1.files.toSet.equals(dir2.files.toSet) &&
|
||||
dir1.subDirs.size.equals(dir2.subDirs.size) &&
|
||||
dir1.subDirs.sortBy(_.name).zip(dir2.subDirs.sortBy(_.name)).forall{
|
||||
case (d1, d2) => directoryEquals(d1, d2)
|
||||
}
|
||||
dir1.files.toSet.equals(dir2.files.toSet) &&
|
||||
dir1.subDirs.size.equals(dir2.subDirs.size) &&
|
||||
dir1.subDirs.sortBy(_.name).zip(dir2.subDirs.sortBy(_.name)).forall {
|
||||
case (d1, d2) => directoryEquals(d1, d2)
|
||||
}
|
||||
}
|
||||
|
||||
// Using `directoryPath`, create a Directory tree starting from root and ending at
|
||||
// `leafDirectory`.
|
||||
private def createDirectory(directoryPath: Path, leafDirectory: Directory): Directory = {
|
||||
TestUtils.splitPath(directoryPath.getParent).foldLeft(leafDirectory) {
|
||||
(accum, name) =>
|
||||
Directory(name, Seq(), Seq(accum))
|
||||
TestUtils.splitPath(directoryPath.getParent).foldLeft(leafDirectory) { (accum, name) =>
|
||||
Directory(name, Seq(), Seq(accum))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -417,8 +417,8 @@ class IndexManagerTest extends HyperspaceSuite with SQLHelper {
|
|||
// Check emitted events.
|
||||
MockEventLogger.emittedEvents match {
|
||||
case Seq(
|
||||
OptimizeActionEvent(_, _, "Operation started."),
|
||||
OptimizeActionEvent(_, _, msg)) =>
|
||||
OptimizeActionEvent(_, _, "Operation started."),
|
||||
OptimizeActionEvent(_, _, msg)) =>
|
||||
assert(
|
||||
msg.contains(
|
||||
"Optimize aborted as no optimizable index files smaller than 1 found."))
|
||||
|
|
@ -477,8 +477,8 @@ class IndexManagerTest extends HyperspaceSuite with SQLHelper {
|
|||
// Check emitted events.
|
||||
MockEventLogger.emittedEvents match {
|
||||
case Seq(
|
||||
OptimizeActionEvent(_, _, "Operation started."),
|
||||
OptimizeActionEvent(_, _, msg)) =>
|
||||
OptimizeActionEvent(_, _, "Operation started."),
|
||||
OptimizeActionEvent(_, _, msg)) =>
|
||||
assert(
|
||||
msg.contains(
|
||||
"Optimize aborted as no optimizable index files smaller than 268435456 found."))
|
||||
|
|
@ -751,16 +751,16 @@ class IndexManagerTest extends HyperspaceSuite with SQLHelper {
|
|||
case Some(s) =>
|
||||
val relations = df.queryExecution.optimizedPlan.collect {
|
||||
case LogicalRelation(
|
||||
HadoopFsRelation(
|
||||
location: PartitioningAwareFileIndex,
|
||||
HadoopFsRelation(
|
||||
location: PartitioningAwareFileIndex,
|
||||
_,
|
||||
dataSchema,
|
||||
_,
|
||||
fileFormat,
|
||||
options),
|
||||
_,
|
||||
dataSchema,
|
||||
_,
|
||||
fileFormat,
|
||||
options),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
_) =>
|
||||
val files = location.allFiles
|
||||
val sourceDataProperties =
|
||||
Hdfs.Properties(Content.fromLeafFiles(files, fileIdTracker).get)
|
||||
|
|
|
|||
|
|
@ -39,12 +39,7 @@ class IndexTest extends SparkFunSuite {
|
|||
|
||||
val entry = IndexLogEntry(
|
||||
config.indexName,
|
||||
CoveringIndex(
|
||||
config.indexedColumns,
|
||||
config.includedColumns,
|
||||
schema,
|
||||
numBuckets,
|
||||
Map()),
|
||||
CoveringIndex(config.indexedColumns, config.includedColumns, schema, numBuckets, Map()),
|
||||
Content(Directory(path)),
|
||||
Source(SparkPlan(sourcePlanProperties)),
|
||||
Map())
|
||||
|
|
|
|||
|
|
@ -154,8 +154,8 @@ class RefreshIndexNestedTest extends QueryTest with HyperspaceSuite {
|
|||
// Check emitted events.
|
||||
MockEventLogger.emittedEvents match {
|
||||
case Seq(
|
||||
RefreshIncrementalActionEvent(_, _, "Operation started."),
|
||||
RefreshIncrementalActionEvent(_, _, msg)) =>
|
||||
RefreshIncrementalActionEvent(_, _, "Operation started."),
|
||||
RefreshIncrementalActionEvent(_, _, msg)) =>
|
||||
assert(msg.contains("Refresh incremental aborted as no source data change found."))
|
||||
case _ => fail()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -152,8 +152,8 @@ class RefreshIndexTest extends QueryTest with HyperspaceSuite {
|
|||
// Check emitted events.
|
||||
MockEventLogger.emittedEvents match {
|
||||
case Seq(
|
||||
RefreshIncrementalActionEvent(_, _, "Operation started."),
|
||||
RefreshIncrementalActionEvent(_, _, msg)) =>
|
||||
RefreshIncrementalActionEvent(_, _, "Operation started."),
|
||||
RefreshIncrementalActionEvent(_, _, msg)) =>
|
||||
assert(msg.contains("Refresh incremental aborted as no source data change found."))
|
||||
case _ => fail()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -138,10 +138,10 @@ class ScoreBasedIndexPlanOptimizerTest extends QueryTest with HyperspaceSuite {
|
|||
private def getAllRootPaths(optimizedPlan: LogicalPlan): Seq[Path] = {
|
||||
optimizedPlan.collect {
|
||||
case LogicalRelation(
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
HadoopFsRelation(location: InMemoryFileIndex, _, _, _, _, _),
|
||||
_,
|
||||
_,
|
||||
_) =>
|
||||
location.rootPaths
|
||||
}.flatten
|
||||
}
|
||||
|
|
|
|||
|
|
@ -116,7 +116,6 @@ class JoinIndexRankerTest extends HyperspaceRuleSuite with SQLHelper {
|
|||
val r_20 = createIndexLogEntry("r2", Seq(t2c1), Seq(t2c2), rightPlan, 20, fileList1, false)
|
||||
setCommonSourceSizeInBytesTag(r_20, rightPlan, fileList1)
|
||||
|
||||
|
||||
val indexPairs = Seq((l_10, r_10), (l_10, r_20), (l_20, r_20))
|
||||
val expectedOrder = Seq((l_20, r_20), (l_10, r_10), (l_10, r_20))
|
||||
val actualOrder = JoinIndexRanker.rank(spark, leftPlan, rightPlan, indexPairs)
|
||||
|
|
|
|||
|
|
@ -138,7 +138,8 @@ class FilterIndexRuleTest extends HyperspaceRuleSuite {
|
|||
}
|
||||
}
|
||||
|
||||
test("Verify FilterIndex rule does not apply if filter does not contain first indexed column.") {
|
||||
test(
|
||||
"Verify FilterIndex rule does not apply if filter does not contain first indexed column.") {
|
||||
val filterCondition =
|
||||
And(IsNotNull(c2), EqualTo(c2, Literal("RGUID_VALUE"))) // c2 is not first indexed column
|
||||
val filterNode = Filter(filterCondition, scanNode)
|
||||
|
|
|
|||
|
|
@ -259,7 +259,8 @@ class JoinIndexRuleTest extends HyperspaceRuleSuite with SQLHelper {
|
|||
}
|
||||
}
|
||||
|
||||
test("Join rule does not update plan if index doesn't satisfy included columns from any side.") {
|
||||
test(
|
||||
"Join rule does not update plan if index doesn't satisfy included columns from any side.") {
|
||||
val t1FilterNode = Filter(IsNotNull(t1c1), t1ScanNode)
|
||||
val t2FilterNode = Filter(IsNotNull(t2c1), t2ScanNode)
|
||||
|
||||
|
|
|
|||
|
|
@ -39,12 +39,7 @@ class JsonUtilsTest extends SparkFunSuite {
|
|||
|
||||
val index = IndexLogEntry(
|
||||
"myIndex",
|
||||
CoveringIndex(
|
||||
Seq("id"),
|
||||
Seq("name", "school"),
|
||||
schema,
|
||||
10,
|
||||
Map()),
|
||||
CoveringIndex(Seq("id"), Seq("name", "school"), schema, 10, Map()),
|
||||
Content(Directory("path")),
|
||||
Source(SparkPlan(sourcePlanProperties)),
|
||||
Map())
|
||||
|
|
|
|||
Загрузка…
Ссылка в новой задаче