sezruby commented on code in PR #13554:
URL: https://github.com/apache/gluten/pull/13554#discussion_r4484031547
##########
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:
This only looks at top-level fields, but lance-spark keeps the blob and JSON
markers on nested fields too, and the native scan returns those nested values
in their stored form as well. With lance-spark 0.8.0, a legacy blob inside a
struct (`info struct<name string, blob binary>` with `lance-encoding:blob` on
`blob`, file format 2.1) passes the check. So does a JSON field inside a struct
or inside a list of structs. Export then throws on its schema check: `Declared
[id Int(64, true) null, info Struct null] but the native scan produced [id
Int(64, true) null, info Struct null]`. Could this walk struct fields, array
elements and map keys and values, checking each field's metadata
(`BlobUtils.isBlobSparkMetadata` / `isBlobV2SparkMetadata`,
`JsonUtils.hasJsonMetadata`)? A blob v2 inside a struct already falls back
here, because its mapped field isn't wrappable.
##########
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:
An int-element vector or a vector of vectors passes this check but still
fails at export. lance-spark 0.8.0's `toArrowSchema` restores a `FixedSizeList`
only for float/double elements (`VectorUtils.shouldBeFixedSizeList`). So a
`FixedSizeList(4)` of int8 maps to `List`, which is wrappable, while the native
scan returns `FixedSizeList(4)`: `Declared [id Int(32, true) null, v List null]
but the native scan produced [id Int(32, true) null, v FixedSizeList(4) null]`.
The same happens for int32 vectors and for a `FixedSizeList(2)` of float
`FixedSizeList(2)`. Both show in the Spark schema: the field carries
`arrow.fixed-size-list.size`, but its mapped field isn't a `FixedSizeList`.
Checking that at any depth would make these fall back, and they would offload
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:
One more case for this list: a microsecond timestamp with a time zone other
than UTC, e.g. `Timestamp(us, America/New_York)`. lance-spark reads it as
`TimestampType` without the zone, and both this check and the export map it
back to `Timestamp(MICROSECOND, UTC)`. So export throws: `Declared [id Int(32,
true) null, v Timestamp(MICROSECOND, UTC) null] but the native scan produced
[id Int(32, true) null, v Timestamp(MICROSECOND, America/New_York) null]`.
--
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]
