malinjawi commented on code in PR #13554:
URL: https://github.com/apache/gluten/pull/13554#discussion_r4492342707
##########
backends-velox/src-lance/main/scala/org/apache/gluten/execution/LanceScanTransformer.scala:
##########
@@ -105,14 +110,74 @@ object LanceScanTransformer {
* rejects both at runtime (checkNoPushedAggregation /
checkNativeSchemaMatchesPartition), but by
* then the node is already offloaded to Velox, so the throw fails the query
instead of falling
* back. Detecting them here keeps such scans on vanilla Spark (e.g. the
fast COUNT(*) metadata
- * reader) rather than offloading a scan that can only crash.
+ * reader) rather than offloading a scan that can only crash. The same holds
for the columns
+ * checked by [[hasExportableColumns]].
*/
def isExportable(batchScan: BatchScanExec): Boolean =
batchScan.inputPartitions.forall {
case p: LanceInputPartition =>
!p.getPushedAggregation.isPresent && p.getReadOptions.getFullTextQuery
== null
case _ => false
+ } && hasExportableColumns(batchScan.scan.readSchema())
+
+ /**
+ * Whether the native fragment scan returns the columns of `schema` as
declared, each as an Arrow
+ * vector that Spark's [[ArrowColumnVector]], which each imported vector is
wrapped in, reads as
+ * the declared type.
+ *
+ * [[LanceArrowStreamScanner.export]] streams the native scan as is, and
only if its schema equals
+ * the one lance-spark's Spark-to-Arrow mapping ([[LanceArrowUtils]])
derives from the declared
+ * schema. The native scan differs for a scan that reads no columns, where
it returns `_rowid`,
+ * and for the columns in [[needsLanceReader]]. Every other column arrives
with its mapped type,
+ * with vectors rewritten as Lists ([[FixedSizeLists]]), and
ArrowColumnVector has no accessor for
+ * some of these types, e.g. FixedSizeBinary, Date(MILLISECOND) and the
elements of a Float16
+ * vector. All of these fail at execution, after the scan was offloaded. A
column's type is
+ * checked by wrapping an empty vector of its field as the scan emits it,
which resolves the type
+ * as the imported batches will.
+ */
+ private def hasExportableColumns(schema: StructType): Boolean =
+ schema.nonEmpty && !schema.fields.exists(needsLanceReader) &&
+ // The mapping throws for types it rejects (e.g. Float16 without Arrow's
Float2Vector), and
+ // the export runs the same mapping.
+ Try(LanceArrowUtils.toArrowSchema(schema, "UTC", true)).toOption.exists {
+ arrowSchema =>
+ schema.fields.zip(arrowSchema.getFields.asScala).forall {
+ case (field, arrowField) =>
+ isWrappable(FixedSizeLists.toListField(arrowField),
field.dataType)
+ }
+ }
+
+ /**
+ * Whether only lance-spark's own reader returns `field` as declared, by
fixing up the native scan
+ * after import. The native scan returns blob columns as blob descriptions
alongside `_rowaddr`,
+ * and JSON columns as JSONB bytes. Lance's metadata columns, and the blob
position and size
+ * columns, are left out of the native scan's projection: lance-spark
requests, computes or
+ * reorders them itself.
+ */
+ private def needsLanceReader(field: StructField): Boolean =
Review Comment:
Good catch, thanks! I missed that lance-spark keeps those markers on nested
fields too. Fixed in c32f49a41: the check now looks inside structs, lists and
maps, and falls back if it finds a blob or JSON field anywhere. I added tests
for JSON in a struct, in a list of structs and as a map value, and for a blob
inside a struct.
##########
backends-velox/src-lance/main/scala/org/apache/gluten/execution/LanceScanTransformer.scala:
##########
@@ -105,14 +110,74 @@ object LanceScanTransformer {
* rejects both at runtime (checkNoPushedAggregation /
checkNativeSchemaMatchesPartition), but by
* then the node is already offloaded to Velox, so the throw fails the query
instead of falling
* back. Detecting them here keeps such scans on vanilla Spark (e.g. the
fast COUNT(*) metadata
- * reader) rather than offloading a scan that can only crash.
+ * reader) rather than offloading a scan that can only crash. The same holds
for the columns
+ * checked by [[hasExportableColumns]].
*/
def isExportable(batchScan: BatchScanExec): Boolean =
batchScan.inputPartitions.forall {
case p: LanceInputPartition =>
!p.getPushedAggregation.isPresent && p.getReadOptions.getFullTextQuery
== null
case _ => false
+ } && hasExportableColumns(batchScan.scan.readSchema())
+
+ /**
+ * Whether the native fragment scan returns the columns of `schema` as
declared, each as an Arrow
+ * vector that Spark's [[ArrowColumnVector]], which each imported vector is
wrapped in, reads as
+ * the declared type.
+ *
+ * [[LanceArrowStreamScanner.export]] streams the native scan as is, and
only if its schema equals
+ * the one lance-spark's Spark-to-Arrow mapping ([[LanceArrowUtils]])
derives from the declared
+ * schema. The native scan differs for a scan that reads no columns, where
it returns `_rowid`,
+ * and for the columns in [[needsLanceReader]]. Every other column arrives
with its mapped type,
+ * with vectors rewritten as Lists ([[FixedSizeLists]]), and
ArrowColumnVector has no accessor for
+ * some of these types, e.g. FixedSizeBinary, Date(MILLISECOND) and the
elements of a Float16
+ * vector. All of these fail at execution, after the scan was offloaded. A
column's type is
+ * checked by wrapping an empty vector of its field as the scan emits it,
which resolves the type
+ * as the imported batches will.
+ */
+ private def hasExportableColumns(schema: StructType): Boolean =
+ schema.nonEmpty && !schema.fields.exists(needsLanceReader) &&
+ // The mapping throws for types it rejects (e.g. Float16 without Arrow's
Float2Vector), and
+ // the export runs the same mapping.
+ Try(LanceArrowUtils.toArrowSchema(schema, "UTC", true)).toOption.exists {
+ arrowSchema =>
+ schema.fields.zip(arrowSchema.getFields.asScala).forall {
+ case (field, arrowField) =>
+ isWrappable(FixedSizeLists.toListField(arrowField),
field.dataType)
Review Comment:
Makes sense, done in c32f49a41. If a field has `arrow.fixed-size-list.size`
but doesn't map back to a `FixedSizeList`, the scan now falls back, at any
depth. I tested an int8 vector and a vector of vectors, written with lance-core
since lance-spark can't write them. And as you said, they'll start offloading
again on their own once lance-spark maps them as `FixedSizeList`.
##########
docs/backends/velox/get-started/VeloxLance.md:
##########
@@ -82,3 +86,6 @@ engages automatically whenever lance-spark is present on the
classpath.
(in the Lance reader) before the Arrow stream is exported, not inside Velox.
- Rare schemas that require JVM post-processing (for example `_rowid` /
`_rowaddr` / blob columns)
are not offloaded.
+- With lance-spark 0.8.0, a scan over an unsigned integer, `LargeList`, time,
or non-microsecond
Review Comment:
Good one, added it to the list. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
