[FEATURE REQUEST]: It would be very nice to use index when underlying table is applied with CACHE TABLE

[SynapsePOC]

Consider this script:

import com.microsoft.hyperspace._
import com.microsoft.hyperspace.Hyperspace
import com.microsoft.hyperspace.index.LogicalPlanSignatureProvider
import com.microsoft.hyperspace.index.IndexConfig
import org.apache.spark.{SparkContext, SparkException}
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.expressions.Window 
import org.apache.spark.sql.expressions.WindowSpec
import scala.util.{Failure, Success, Try}
import com.microsoft.hyperspace.actions.Constants
import com.microsoft.hyperspace.util.{FileUtils, LogicalPlanUtils, ResolverUtils}
import org.apache.hadoop.fs.Path

val pipelineSpark :SparkSession = SparkSession.builder().config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension").config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog").getOrCreate()
import pipelineSpark.implicits._

pipelineSpark.enableHyperspace()
val hyperspace = new Hyperspace(pipelineSpark)

val ws = Window.orderBy("C1")
val df = pipelineSpark.read.json("testJson.json").withColumn("RowId", row_number() over ws)

val path = new Path("delta")
FileUtils.delete(path,true)
val path = new Path("spark-warehouse/indexes")
FileUtils.delete(path,true)
df.write.format("parquet").save("delta")
val df = pipelineSpark.read.format("parquet").load("delta")
df.createOrReplaceGlobalTempView("jsont")
val indexConfig = new IndexConfig("ix1", Seq("C1"), Seq("C2"))
hyperspace.createIndex(df, indexConfig)
val t = pipelineSpark.sql("SELECT C1, C2 FROM global_temp.jsont WHERE C1=1")
hyperspace.explain(t, true)
pipelineSpark.sql("CACHE TABLE global_temp.jsont")
val t = pipelineSpark.sql("SELECT C1, C2 FROM global_temp.jsont WHERE C1=1")
hyperspace.explain(t, true)

The last explain produces plan as following:

scala> hyperspace.explain(t, true)
=============================================================
Plan with indexes:
=============================================================
Filter (isnotnull(C1#30L) && (C1#30L = 1))
+- InMemoryTableScan [C1#30L, C2#31L], [isnotnull(C1#30L), (C1#30L = 1)]

=============================================================
Plan without indexes:
=============================================================
Filter (isnotnull(C1#30L) && (C1#30L = 1))
+- InMemoryTableScan [C1#30L, C2#31L], [isnotnull(C1#30L), (C1#30L = 1)]

=============================================================
Indexes used:
=============================================================

=============================================================
Physical operator stats:
=============================================================

In my mind this is truly reduces index usage scenario set. It would be very nice if caching table actually promoted index use, at the moment caching index actually turns index use off.
I think one of the issues logged already, can help to address it
#30
but I would like to make sure this is not lost in translation.

[imback82]

Thanks @SynapsePOC for reporting this. AFAIK, caching the base table doesn't make sense in Hyperspace, since our index can have a subset of the original columns. (but I see that it could be beneficial if we could cache the index directly via some mechanism, but we need to investigate more on this.)

You could always cache the result of query:

val t = spark.sql("SELECT C1, C2 FROM tbl WHERE C1=1")
t.cache

, in which case, cache can be populated with index, but I don't think this is the scenario you are targeting.

cc-ing few people if I missed anything: @rapoth @apoorvedave1

[SynapsePOC]

+ Prasad From: Terry Kim <notifications@github.com> Sent: Saturday, August 22, 2020 12:45 PM To: microsoft/hyperspace <hyperspace@noreply.github.com> Cc: Miron Berlin (People Tech Group) <v-bemiro@microsoft.com>; Mention <mention@noreply.github.com> Subject: Re: [microsoft/hyperspace] [PROPOSAL]: It would be very nice to use index when underlying table is applied with CACHE TABLE (#130) Thanks @SynapsePOC<https://nam06.safelinks.protection.outlook.com/?url=https%3A%2F%2Fgithub.com%2FSynapsePOC&data=02%7C01%7Cv-bemiro%40microsoft.com%7C0e44bb34cc574070976a08d846d3e953%7C72f988bf86f141af91ab2d7cd011db47%7C1%7C0%7C637337223254777601&sdata=eagocSIA7%2Fk8x7EWkfDqqakAUzYYlo1ooQaUKczEPKE%3D&reserved=0> for reporting this. AFAIK, caching the base table doesn't make sense in Hyperspace, since our index can have a subset of the original columns. (but I see that it could be beneficial if we could cache the index directly via some mechanism, but we need to investigate a bit more to see how we can preserve bucketing, etc.) You could always cache the result of query: val t = spark.sql("SELECT C1, C2 FROM tbl WHERE C1=1") t.cache , in which case, cache can be populated with index, but I don't think this is the scenario you are targeting. cc-ing few people if I missed anything: @rapoth<https://nam06.safelinks.protection.outlook.com/?url=https%3A%2F%2Fgithub.com%2Frapoth&data=02%7C01%7Cv-bemiro%40microsoft.com%7C0e44bb34cc574070976a08d846d3e953%7C72f988bf86f141af91ab2d7cd011db47%7C1%7C0%7C637337223254787595&sdata=1aJNdn8Qu%2B6LIIYCJffk4fKXtgQtljbvkerk0mBYpvw%3D&reserved=0> @apoorvedave1<https://nam06.safelinks.protection.outlook.com/?url=https%3A%2F%2Fgithub.com%2Fapoorvedave1&data=02%7C01%7Cv-bemiro%40microsoft.com%7C0e44bb34cc574070976a08d846d3e953%7C72f988bf86f141af91ab2d7cd011db47%7C1%7C0%7C637337223254787595&sdata=v33JLi7iMQd1K5HPFSnfIp9UoUix0Syn2IqwD6RjtRg%3D&reserved=0> — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub<https://nam06.safelinks.protection.outlook.com/?url=https%3A%2F%2Fgithub.com%2Fmicrosoft%2Fhyperspace%2Fissues%2F130%23issuecomment-678684485&data=02%7C01%7Cv-bemiro%40microsoft.com%7C0e44bb34cc574070976a08d846d3e953%7C72f988bf86f141af91ab2d7cd011db47%7C1%7C0%7C637337223254797590&sdata=EKazbH7G0w35uWmlJzNjzq3L0OLyWw8TFmP%2BEJWY1cw%3D&reserved=0>, or unsubscribe<https://nam06.safelinks.protection.outlook.com/?url=https%3A%2F%2Fgithub.com%2Fnotifications%2Funsubscribe-auth%2FAQL4U4DO2HNBEOGPWGT2OU3SCAN5HANCNFSM4QGJHF3A&data=02%7C01%7Cv-bemiro%40microsoft.com%7C0e44bb34cc574070976a08d846d3e953%7C72f988bf86f141af91ab2d7cd011db47%7C1%7C0%7C637337223254797590&sdata=euUTOfnO1RrxxoFN3oFfEgGCDQHnIVYoIX%2FGeJI4zCY%3D&reserved=0>.

[rapoth]

We will investigate on what is required and whether this is feasible and get back. Thanks!

[SynapsePOC]

The value of this proposal can go manyfold, if cache is co-located with partition, thus retaining in the hands of user control over distribution of data amongst the nodes, participating in cluster. Thus, providing for the applications requiring most intensive query activity, to harvest significant volumes of data, without having to deal with shuffling and limitations of the driver node.
In that regard the ability to co - partition indexes along with data can present a huge plus, for users, who partition their data neatly.
Yes, if we try to seek corner cases, it can potentially hurt users, where data is askew amongst the nodes, but this is a separate issue.

[sezruby]

For the scenario in the description, the problem is that Spark's Cache Manager changed "Relation" of cached df to "InMemoryTableScan" before the optimizer which applies indexes if possible.
In that case, we don't replace the plan to use the index (uncached) - and I don't think we have to; it is not desirable in general.

These are some possible scenarios:

1. caching "index data" itself, instead of the source table.

• whole index data will be cached.
• we can apply the cached index for all query that can be applied the index.
  e.g

hyperspace.cache("indexName", ...)

(new api change required #254)

2. caching query result using "index data"

• or we could do cache index data with Filter or Join conditions:

val filter = df.filter("C1 = 1")
filter.cache

val t = spark.sql("SELECT C1, C2 FROM tbl WHERE C1=1")
t.cache

In this way, we could apply index data while using caching.
To do this, hyperspace should be enabled at cache time.

Or would you still like to utilize "cached" table along with "cached" index data though inefficient memory footprint?
@SynapsePOC