Skip to content
This repository has been archived by the owner on Jan 3, 2023. It is now read-only.

[SQL-DS-CACHE-38][POAE7-892]update Plasma dependency #39

Merged
merged 3 commits into from
Mar 12, 2021
Merged

[SQL-DS-CACHE-38][POAE7-892]update Plasma dependency #39

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8dffa8d
use standard Plasma API
jikunshang Mar 11, 2021
6ce2375
add comments in pom file to illustrate dependency.
jikunshang Mar 11, 2021
20b4885
fix typo
jikunshang Mar 11, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion Plasma-based-cache/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
</parent>
<artifactId>plasma-sql-ds-cache</artifactId>
<version>1.1.0</version>
<name>Plasma Based SQL DS Cacahe</name>
<name>Plasma Based SQL DS Cache</name>
<packaging>jar</packaging>


Expand Down Expand Up @@ -274,6 +274,10 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
<!--
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@winningsix modified arrow-plasma-3.0.0.jar will not update to mvn central repo. For build purpose, I'd like to keep it here.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How does it work when we release this?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

if user want to build by themself, they still use this one. Otherwise user could use jar or so file provided by conda.

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Got you. So basically, Haifeng team will build one version with Arrow-3.0 and in source code we still use old version. Right?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes.

We've changed something based on Apache Arrow. For mvn build, will use this dependency
since it's been published on central maven repo.
-->
<dependency>
<groupId>com.intel.arrow</groupId>
<artifactId>arrow-plasma</artifactId>
Expand Down
21 changes: 10 additions & 11 deletions ...he/src/main/scala/org/apache/spark/sql/execution/datasources/oap/filecache/OapCache.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -996,7 +996,7 @@ class ExternalCache(fiberType: FiberType) extends OapCache with Logging {
val objectId = hash(fiberId.toString)
val plasmaClient = plasmaClientPool(clientRoundRobin.getAndAdd(1) % clientPoolSize)
try {
val buf: ByteBuffer = plasmaClient.create(objectId, fiberLength.toInt)
val buf: ByteBuffer = plasmaClient.create(objectId, fiberLength.toInt, null)
ExternalDataFiber(buf, objectId, plasmaClient)
}
catch {
Expand Down Expand Up @@ -1065,19 +1065,18 @@ class ExternalCache(fiberType: FiberType) extends OapCache with Logging {
val objectId = hash(fiberId.toString)
if(contains(fiberId)) {
var fiberCache : FiberCache = null
try{
logDebug(s"Cache hit, get from external cache.")
val plasmaClient = plasmaClientPool(clientRoundRobin.getAndAdd(1) % clientPoolSize)
val buf: ByteBuffer = plasmaClient.getObjAsByteBuffer(objectId, -1, false)
logDebug(s"Cache hit, get from external cache.")
val plasmaClient = plasmaClientPool(clientRoundRobin.getAndAdd(1) % clientPoolSize)
val buf: ByteBuffer = plasmaClient.getObjAsByteBuffer(objectId, -1, false)
if(buf.asInstanceOf[DirectBuffer].address() == 0) {
logWarning("Get return an invalid value.")
fiberCache = cache(fiberId)
cacheMissCount.addAndGet(1)
} else {
cacheHitCount.addAndGet(1)
fiberCache = ExternalDataFiber(buf, objectId, plasmaClient)
}
catch {
case getException : plasma.exceptions.PlasmaGetException =>
logWarning("Get exception: " + getException.getMessage)
fiberCache = cache(fiberId)
cacheMissCount.addAndGet(1)
}

fiberCache.fiberId = fiberId
fiberCache.occupy()
cacheGuardian.addRemovalFiber(fiberId, fiberCache)
Expand Down