[GLUTEN-4241][VL] Add plan node to convert Vanilla spark columnar format data to Velox columnar format data #4818
Conversation
|
Run Gluten Clickhouse CI |
|
Run Gluten Clickhouse CI |
|
|
||
| ArrowSchema cSchema; | ||
| ArrowArray arrowArray; | ||
| exportToArrow(vector, cSchema, ArrowUtils::getBridgeOptions()); |
There was a problem hiding this comment.
If I'm understanding well, there might be a bug that velox doesn't handle short decimal well when converting arrowArray-> rowVector
In exportToArrow, we can see it needs to convert to int128_t since it's arrow short decimal.
rows.apply([&](vector_size_t i) {
int128_t value = buf.as<int64_t>()[i];
memcpy(dst + (j++) * sizeof(int128_t), &value, sizeof(int128_t));
});But in importFromArrow, velox directly handle it as int64_t, whereas this test won't pass.
values = wrapInBufferView(
arrowArray.buffers[1], arrowArray.length * type->cppSizeInBytes());
There was a problem hiding this comment.
I see, let me try to fix velox firstly
There was a problem hiding this comment.
fix it here: facebookincubator/velox#8957
|
Run Gluten Clickhouse CI |
There was a problem hiding this comment.
Thanks! I didn't take look in the PR in detail but overall the direction looks promising.
Also, would you like to add a little bit more explanation in PR description ? Like it supports all datatypes or just some of them, it will be turned on or off by default, what's the config key to turn it on/off, etc. That will help further users getting started with the feature quickly.
| new ValidatorApiImpl().doSchemaValidate(schema).foreach { | ||
| reason => | ||
| throw new UnsupportedOperationException( | ||
| s"Input schema contains unsupported type when performing columnar" + | ||
| s" to columnar for $schema " + s"due to $reason") | ||
| } |
There was a problem hiding this comment.
It looks a little bit weird to do validation during execution. Would you like to share the reason doing this?
There was a problem hiding this comment.
Yea, I follow RowToVeloxColumnarExec to do the validation here, you're right, it looks weird to do so during execution, I also have the same doubt when seeing RowToVeloxColumnarExec...
Let me fix it by overriding doValidateInternal instead.
gluten-core/src/main/scala/io/glutenproject/extension/ColumnarOverrides.scala
Outdated
Show resolved
Hide resolved
| case p: RowToColumnarExecBase if p.child.isInstanceOf[ColumnarToRowExec] => | ||
| val replacedChild = replaceWithColumnarToColumnar( | ||
| p.child.asInstanceOf[ColumnarToRowExec].child) | ||
| BackendsApiManager.getSparkPlanExecApiInstance.genColumnarToColumnarExec(replacedChild) |
There was a problem hiding this comment.
The case condition says it matches vanilla-columnar->row->gluten-columnar but it calls a general interface to do columnar->columnar. Do we need to specialize the interface name to make sure we exactly request for vanilla-columnar->gluten-columnar transition here?
There was a problem hiding this comment.
The case condition here tries to match rowTocolumnar -> ColumnarToRow, for e.g.
Filter
RowToVeloxColumnar
ColumnarToRow
FileScan xxWe tries to match RowToVeloxColumnar->ColumnarToRow here with ColumnarToColumnar
There was a problem hiding this comment.
Thanks for the explanation.
If it's for ColumnarToRow + RowToVeloxColumnar,
Would it be better to rename the method genColumnarToColumnarExec with genVanillaColumnarToGlutenColumnarExec or something?
There was a problem hiding this comment.
So it's clearer that we are not doing Gluten -> Vanilla c2c transition here.
|
Do you convert to Velox format directly? or convert to Arrow then to Velox? |
Convert to Arrow firstly, then to velox |
make sense. We may upstream the parquet columnar format => arrow format to Spark |
|
Run Gluten Clickhouse CI |
|
Run Gluten Clickhouse CI |
| if (arrowArray != null) { | ||
| arrowArray.release() | ||
| arrowArray.close() | ||
| arrowConverter.reset() | ||
| arrowArray = null | ||
| } |
There was a problem hiding this comment.
Seems to be that the file includes a bunch of manual life cycle management calls like _.close _.release _.reset, etc. Would you like to review them to see if we can eliminate some? I recall that recycleIterator / recyclePayloads can do some of the clean up jobs automatically.
It's a non-blocking comment so feel free to do that in another PR. But anyway we'd try best to make code here easier to maintain from further developers.
There was a problem hiding this comment.
We need to reset arrowArray and arrowConverter when iterate each item, so recyclePayload is reasonable, but it only passes the result of the next(), which is ColumnarBatch, we cannot pass arrowArray to close them.
So here instead I add an extra method releaseArrowBuf to release them to reduce duplicates.
|
Run Gluten Clickhouse CI |
|
Run Gluten Clickhouse CI |
|
@zhztheplayer can you check how the memory is allocated during the conversion? Where the arrow memory is allocated? how many memcpy during the conversion? Is there onheap=>offheap copy? |
@boneanxs If you'd like to address the questions also, thanks. I believe the patch
Also would be great if you could share thoughts about the risk of memory leaks this approach may bring @boneanxs . Overall the PR's writing looks fine to me and we had removed most of the unsafe APIs but still there might be some. Let's check this part carefully too. |
|
Let's document the conversion clearly here. I have a impression that parquet-mr can take use of offheap memory for columnar data. If so the best case is that we can avoid any memcpy during the conversion (not considered about arrow=>velox conversion). But we will needs some work to avoid it. If parquet scan is onheap, we can't avoid the onheap->offheap copy, but then we should reuse the offheap buffer. If we can't make sure how the conversion happens, let's use memcpy as long we need, make sure it's 100% no memleak for now. |
Hey, @FelixYBW for each columnar batch, before the conversion, this pr tries to allocate offheap memory to perform So yes, there will be one memcpy from In native side, I simply uses
Sry @zhztheplayer, I might miss something, are you saying
The extra memory here acquired are
I'm not sure whether parquet-mr can directly use offheap memory, but spark can export parquet data as offheap columnar batch by enabling I can also try to do benchmarks comparing with VanillarColumnarBatchToRow->RowToVeloxColumnar if necessary, I think at least for memcpy, |
Ahh then it's fine enough. Some of the code is so similar to Gluten's |
Thank you for explanation. You may try to enable spark.sql.columnVector.offheap.enabled. onheap to offheap memcpy is more expensive than offheap to offheap. Do you know how ArrowFieldWriter allocate memory? is it from unsafe API or direct memory? @zhztheplayer Do you know? |
|
Hey @FelixYBW for example, IntVector ValueBuffer is allocated from arrow |
|
I think it's OK to have it disabled by default. @boneanxs Can you add a CI case for the feature to run TPC-H / TPC-DS tests? Example: You can use gluten-it arg Also if And sorry for late response. Let's have this merged asap. |
|
Oh, just noted the PR is still open and have many conflict. @boneanxs would you like to continue? |
|
Hey @FelixYBW @zhztheplayer, yea, I'm willing to continue this pr. After last comment, I actually have run some benchmarks in my local environment, and found there's no obviously improvement comparing VanillaColumnar->row->VeloxColumnar(Also I notice there's a compatible issue since this pr relies on spark arrow version, which is conflict with gluten). I'm still checking why this(performance not improved) could happen(but didn't catch time recently), will update here once I'm done other things. |
|
@boneanxs hi, could you please help to do rebase? There's a big commit(renaming io.glutneproject to org.apache.gluten) when we migrate to apache repo. thanks |
|
This PR is stale because it has been open 45 days with no activity. Remove stale label or comment or this will be closed in 10 days. |
|
comment to keep the PR open as it could be a valuable topic |
|
This PR is stale because it has been open 45 days with no activity. Remove stale label or comment or this will be closed in 10 days. |
|
This PR was auto-closed because it has been stalled for 10 days with no activity. Please feel free to reopen if it is still valid. Thanks. |
What changes were proposed in this pull request?
Add plan node to convert Vanilla spark columnar format data to Velox columnar format data
This pr tries to convert from Spark columnar batch -> arrow Array -> velox columnar batch, all primitive types(decimal types requires: facebookincubator/velox#8957) and Map/Array types are supported.
Users can enable it by setting
spark.gluten.sql.columnar.vanillaColumnarToNativeColumnar, which is by default false.(Fixes: #4241)
How was this patch tested?
Added tests