This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch branch-1.2.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/branch-1.2.x by this push:
new 6f8ea45f2d70 fix(variant): port #20041 to branch-1.2.x (#20160)
6f8ea45f2d70 is described below
commit 6f8ea45f2d706051373084f79fee5e859653aac9
Author: voonhous <[email protected]>
AuthorDate: Thu Oct 1 12:25:24 2026 +0800
fix(variant): port #20041 to branch-1.2.x (#20160)
On Spark 4.1, where spark.sql.variant.pushVariantIntoScan is on by
default, the record context builds its row writers from the engine
schema, which types a variant as VariantType, while the rows carry
the projection struct. A query that pushes several paths of which a
row holds few then fails with NegativeArraySizeException on a MOR
read that merges a log file and on a bootstrap read. The
partial-update merge reads the base record through the same types.
Hand-port of master 76be6c65b2c1. Three of the four files apply as
they are. SparkFileFormatInternalRowReaderContext is resolved by
hand: this branch projects log records for every merger, so the row
shape is installed whenever the query carries a projection, without
master's payload-merger exclusion.
TestVariantPushVariantIntoScan gains a wide projection and a
partial-update MERGE INTO on the top-level sweep, and a bootstrap
test. The wide projection and the bootstrap test fail without the
port; the MERGE INTO check fails with the SparkRecordMergingUtils
hunk alone left out.
---
.../apache/hudi/merge/SparkRecordMergingUtils.java | 12 ++-
.../hudi/BaseSparkInternalRecordContext.java | 44 ++++++++-
.../hudi/BaseSparkInternalRowReaderContext.java | 7 +-
.../SparkFileFormatInternalRowReaderContext.scala | 22 ++++-
.../schema/TestVariantPushVariantIntoScan.scala | 107 ++++++++++++++++++++-
5 files changed, 179 insertions(+), 13 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/merge/SparkRecordMergingUtils.java
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/merge/SparkRecordMergingUtils.java
index 29408c131dfa..ebb91f48d717 100644
---
a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/merge/SparkRecordMergingUtils.java
+++
b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/merge/SparkRecordMergingUtils.java
@@ -19,6 +19,7 @@
package org.apache.hudi.merge;
+import org.apache.hudi.BaseSparkInternalRecordContext;
import org.apache.hudi.HoodieSchemaConversionUtils;
import org.apache.hudi.common.engine.RecordContext;
import org.apache.hudi.common.model.HoodieRecordMerger;
@@ -32,6 +33,7 @@ import org.apache.hudi.common.util.collection.Pair;
import org.apache.spark.sql.HoodieInternalRowUtils;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
@@ -119,16 +121,22 @@ public class SparkRecordMergingUtils {
Map<Integer, StructField> mergedIdToFieldMapping =
mergedSchemaPair.getLeft();
Map<String, Integer> oldNameToIdMapping =
getCachedFieldNameToIdMapping(oldSchema);
Map<String, Integer> newPartialNameToIdMapping =
getCachedFieldNameToIdMapping(newSchema);
+ // The values are read in the shape the rows carry: under a
PushVariantIntoScan projection a variant is the
+ // projection struct, which a VariantType read would decode through
UnsafeRow.getVariant.
+ StructType mergedRowStruct = recordContext instanceof
BaseSparkInternalRecordContext
+ ? ((BaseSparkInternalRecordContext)
recordContext).getRowStructType(mergedSchemaPair.getRight().getRight())
+ : mergedSchemaPair.getRight().getLeft();
List<Object> values = new ArrayList<>(mergedIdToFieldMapping.size());
for (int fieldId = 0; fieldId < mergedIdToFieldMapping.size();
fieldId++) {
StructField structField = mergedIdToFieldMapping.get(fieldId);
+ DataType dataType = mergedRowStruct.fields()[fieldId].dataType();
Integer ordInPartialUpdate =
newPartialNameToIdMapping.get(structField.name());
if (ordInPartialUpdate != null) {
// The field exists in the newer record; picks the value from newer
record
- values.add(newPartialRow.get(ordInPartialUpdate,
structField.dataType()));
+ values.add(newPartialRow.get(ordInPartialUpdate, dataType));
} else {
// The field does not exist in the newer record; picks the value
from older record
- values.add(oldRow.get(oldNameToIdMapping.get(structField.name()),
structField.dataType()));
+ values.add(oldRow.get(oldNameToIdMapping.get(structField.name()),
dataType));
}
}
InternalRow mergedRow = new GenericInternalRow(values.toArray());
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRecordContext.java
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRecordContext.java
index 6bd5420af014..53c0c85dbfa2 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRecordContext.java
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRecordContext.java
@@ -46,6 +46,7 @@ import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.UnaryOperator;
import scala.Function1;
@@ -55,6 +56,9 @@ import static
org.apache.spark.sql.HoodieInternalRowUtils.getCachedSchema;
public abstract class BaseSparkInternalRecordContext extends
RecordContext<InternalRow> {
private OrderingValueEngineTypeConverter orderingValueConverter;
+ private UnaryOperator<StructType> rowShape;
+ // The row shape applied per engine schema, so the per-record accessors do
not rebuild the overlay.
+ private final Map<HoodieSchema, StructType> rowStructTypes = new
ConcurrentHashMap<>();
protected BaseSparkInternalRecordContext(HoodieTableConfig tableConfig) {
super(tableConfig, new DefaultJavaTypeConverter());
@@ -73,7 +77,10 @@ public abstract class BaseSparkInternalRecordContext extends
RecordContext<Inter
}
private static Object getFieldValueFromInternalRowInternal(InternalRow row,
HoodieSchema recordSchema, String fieldName, boolean convertToJavaType) {
- StructType structType = getCachedSchema(recordSchema);
+ return getFieldValueFromInternalRowInternal(row,
getCachedSchema(recordSchema), fieldName, convertToJavaType);
+ }
+
+ private static Object getFieldValueFromInternalRowInternal(InternalRow row,
StructType structType, String fieldName, boolean convertToJavaType) {
scala.Option<HoodieUnsafeRowUtils.NestedFieldPath> cachedNestedFieldPath =
HoodieInternalRowUtils.getCachedPosList(structType, fieldName);
if (cachedNestedFieldPath.isDefined()) {
@@ -106,7 +113,7 @@ public abstract class BaseSparkInternalRecordContext
extends RecordContext<Inter
@Override
public Object getValue(InternalRow row, HoodieSchema schema, String
fieldName) {
- return getFieldValueFromInternalRow(row, schema, fieldName);
+ return getFieldValueFromInternalRowInternal(row, getRowStructType(schema),
fieldName, false);
}
@Override
@@ -215,14 +222,43 @@ public abstract class BaseSparkInternalRecordContext
extends RecordContext<Inter
if (internalRow instanceof UnsafeRow) {
return internalRow;
}
- final UnsafeProjection unsafeProjection =
HoodieInternalRowUtils.getCachedUnsafeProjection(schema);
+ final UnsafeProjection unsafeProjection;
+ if (rowShape == null) {
+ unsafeProjection =
HoodieInternalRowUtils.getCachedUnsafeProjection(schema);
+ } else {
+ StructType rowStructType = getRowStructType(schema);
+ unsafeProjection =
HoodieInternalRowUtils.getCachedUnsafeProjection(rowStructType, rowStructType);
+ }
return unsafeProjection.apply(internalRow);
}
+ /**
+ * Installs the Spark type the rows of an engine schema actually carry in
this read; null, the default, means the
+ * plain conversion. {@code SparkFileFormatInternalRowReaderContext}
installs the PushVariantIntoScan overlay here
+ * (see its setSchemaHandler), because every row writer and field accessor
this context builds from an engine
+ * schema has to be typed over that shape: a VariantType-typed one reads a
projection struct through
+ * UnsafeRow.getVariant instead of copying it across. That covers {@link
#projectRecord}, {@link #getValue} (the
+ * partial-update merges read the older record through it) and {@link
#toBinaryRow}; SparkRecordMergingUtils
+ * reads the merged record's fields through {@link #getRowStructType} too.
+ */
+ public void setRowShape(UnaryOperator<StructType> rowShape) {
+ this.rowShape = rowShape;
+ rowStructTypes.clear();
+ }
+
+ /**
+ * The Spark type the rows of {@code schema} carry in this read: the plain
conversion, or the row shape installed
+ * by {@link #setRowShape} when the reader hands its rows over in a
rewritten shape.
+ */
+ public StructType getRowStructType(HoodieSchema schema) {
+ StructType structType = getCachedSchema(schema);
+ return rowShape == null ? structType :
rowStructTypes.computeIfAbsent(schema, s -> rowShape.apply(structType));
+ }
+
@Override
public UnaryOperator<InternalRow> projectRecord(HoodieSchema from,
HoodieSchema to, Map<String, String> renamedColumns) {
Function1<InternalRow, UnsafeRow> unsafeRowWriter =
- HoodieInternalRowUtils.getCachedUnsafeRowWriter(getCachedSchema(from),
getCachedSchema(to), renamedColumns, Collections.emptyMap());
+
HoodieInternalRowUtils.getCachedUnsafeRowWriter(getRowStructType(from),
getRowStructType(to), renamedColumns, Collections.emptyMap());
return row -> (InternalRow) unsafeRowWriter.apply(row);
}
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java
index e4a34ec60acf..5cde7f35cd9b 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/BaseSparkInternalRowReaderContext.java
@@ -44,7 +44,6 @@ import java.util.stream.Collectors;
import scala.Function1;
import static
org.apache.hudi.common.config.HoodieReaderConfig.RECORD_MERGE_IMPL_CLASSES_WRITE_CONFIG_KEY;
-import static org.apache.spark.sql.HoodieInternalRowUtils.getCachedSchema;
/**
* An abstract class implementing {@link HoodieReaderContext} to handle {@link
InternalRow}s.
@@ -81,6 +80,8 @@ public abstract class BaseSparkInternalRowReaderContext
extends HoodieReaderCont
/**
* Constructs a transformation that will take a row and convert it to a new
row with the given schema and adds in the values for the partition columns if
they are missing in the returned row.
* It is assumed that the `to` schema will contain the partition fields.
+ * The data-file rows arrive in the shape {@link #getFileRecordIterator}
read them in, so the writer is typed over
+ * the record context's row shape rather than over the plain engine-schema
conversion.
* @param from the original schema
* @param to the schema the row will be converted to
* @param partitionFieldAndValues the partition fields and their values, if
any are required by the reader
@@ -89,8 +90,10 @@ public abstract class BaseSparkInternalRowReaderContext
extends HoodieReaderCont
protected UnaryOperator<InternalRow> getBootstrapProjection(HoodieSchema
from, HoodieSchema to, List<Pair<String, Object>> partitionFieldAndValues) {
Map<Integer, Object> partitionValuesByIndex =
partitionFieldAndValues.stream()
.collect(Collectors.toMap(pair ->
to.getField(pair.getKey()).orElseThrow(() -> new
IllegalArgumentException("Missing field: " + pair.getKey())).pos(),
Pair::getRight));
+ BaseSparkInternalRecordContext sparkRecordContext =
(BaseSparkInternalRecordContext) recordContext;
Function1<InternalRow, UnsafeRow> unsafeRowWriter =
- HoodieInternalRowUtils.getCachedUnsafeRowWriter(getCachedSchema(from),
getCachedSchema(to), Collections.emptyMap(), partitionValuesByIndex);
+
HoodieInternalRowUtils.getCachedUnsafeRowWriter(sparkRecordContext.getRowStructType(from),
+ sparkRecordContext.getRowStructType(to), Collections.emptyMap(),
partitionValuesByIndex);
return row -> (InternalRow) unsafeRowWriter.apply(row);
}
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
index 1429f5d5363f..a17db4883744 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
@@ -26,6 +26,7 @@ import org.apache.hudi.common.fs.FSUtils
import org.apache.hudi.common.model.{HoodieFileFormat, HoodieRecord}
import org.apache.hudi.common.schema.{HoodieSchema, HoodieSchemaUtils}
import org.apache.hudi.common.table.HoodieTableConfig
+import org.apache.hudi.common.table.read.FileGroupReaderSchemaHandler
import
org.apache.hudi.common.table.read.buffer.PositionBasedFileGroupRecordBuffer.ROW_INDEX_TEMPORARY_COLUMN_NAME
import org.apache.hudi.common.util.{Option => HOption}
import org.apache.hudi.common.util.ValidationUtils.checkState
@@ -102,15 +103,30 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
})
}
+ // Whether the query carries a Spark 4.1 PushVariantIntoScan projection at
all.
+ private lazy val hasVariantProjection: Boolean =
+ sparkRequiredSchema.exists(_.fields.exists(f =>
sparkAdapter.containsVariantProjection(f.dataType)))
+
+ // HoodieFileGroupReader installs the schema handler before it asks for the
output converter, so
+ // this is where the record context learns what shape its rows carry.
Base-file rows are read in
+ // the projected shape and getLogBlockRecordProjection rewrites log rows
into it, so every row
+ // writer the record context builds from an engine schema needs the shape
(#20040).
+ override def setSchemaHandler(schemaHandler:
FileGroupReaderSchemaHandler[InternalRow]): Unit = {
+ super.setSchemaHandler(schemaHandler)
+ if (hasVariantProjection) {
+ val requiredStruct = sparkRequiredSchema.get
+ recordContext.asInstanceOf[BaseSparkInternalRecordContext].setRowShape(
+ overlayVariantProjections(_, requiredStruct))
+ }
+ }
+
// Aligns log-block records with the PushVariantIntoScan-projected variant
shape before
// they reach the merger. Preserves merger metadata cols (_hoodie_record_key,
// _tmp_metadata_row_index) which the merger reads by ordinal — projecting
down to the
// bare required schema would drop them and the merger would read garbage
offsets.
override def getLogBlockRecordProjection(
dataBlockSchema: HoodieSchema): HOption[JFunction[InternalRow,
InternalRow]] = {
- val needsProjection =
- sparkRequiredSchema.exists(_.fields.exists(f =>
sparkAdapter.containsVariantProjection(f.dataType)))
- if (!needsProjection) {
+ if (!hasVariantProjection) {
return HOption.empty[JFunction[InternalRow, InternalRow]]()
}
val req = sparkRequiredSchema.get
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
index 790317b5c86d..a05a1f844cda 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
@@ -18,9 +18,14 @@
package org.apache.spark.sql.hudi.dml.schema
-import org.apache.hudi.{HoodieSparkUtils, SparkAdapterSupport}
+import org.apache.hudi.{DataSourceWriteOptions, HoodieSparkUtils,
SparkAdapterSupport}
+import org.apache.hudi.common.config.HoodieMetadataConfig
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
+import org.apache.hudi.config.{HoodieBootstrapConfig, HoodieWriteConfig}
+import org.apache.hudi.keygen.NonpartitionedKeyGenerator
+import org.apache.hudi.testutils.HoodieClientTestUtils.createMetaClient
+import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
import org.apache.spark.sql.types.{DataType, StructType}
@@ -37,7 +42,8 @@ import scala.util.{Failure, Success, Try}
* (AVRO) and parquet (SPARK) log blocks on the MOR legs.
*
* Every leg runs to the end and its verdict is recorded, so one wrong result
does not hide the
- * others.
+ * others. The wide-projection, partial-update and bootstrap checks are the
regression tests of
+ * #20041 (row writers typed over the projected shape).
*/
class TestVariantPushVariantIntoScan extends HoodieSparkSqlTestBase {
@@ -47,6 +53,15 @@ class TestVariantPushVariantIntoScan extends
HoodieSparkSqlTestBase {
private def merged(id: Int): String = if (id < 3) s"x$id" else if (id == 3)
"y3" else null
private def mergedJson(id: Int): String = Option(merged(id)).map(k =>
s"""{"k":"$k"}""").orNull
+ // Eight pushed paths of which a row holds one, so seven members of the
projection struct are
+ // null. A row writer typed over VariantType instead of the projected shape
reads that struct's
+ // null bitset as a variant length and throws NegativeArraySizeException
(#20040).
+ private def widePaths(column: String): String =
+ (s"try_variant_get($column, '$$.k', 'string')" +:
+ ('b' to 'h').map(c => s"try_variant_get($column, '$$.$c',
'bigint')")).mkString(", ")
+
+ private def wideRow(id: Int): Seq[Any] = Seq(id, merged(id)) ++
Seq.fill(7)(null)
+
private def variantProjectionPushedIntoScan(sql: String): Boolean = {
def containsProjection(dataType: DataType): Boolean = dataType match {
case st: StructType =>
@@ -142,7 +157,21 @@ class TestVariantPushVariantIntoScan extends
HoodieSparkSqlTestBase {
// projected as a struct OF nulls instead of a NULL struct loses this
row.
checkAnswer(s"select id from $tableName where v is null")(Seq(4))
checkAnswer(s"select count(*) from $tableName where v is not
null")(Seq(4))
+ checkAnswer(s"select id, ${widePaths("v")} from $tableName order by id")(
+ (0 until 5).map(wideRow): _*)
assertPushed(s"select id, variant_get(v, '$$.k', 'string') from
$tableName", pushed, tableName)
+ if (tableType == "mor") {
+ // A MERGE INTO that assigns ts alone writes a partial log block on
MOR, so the read rebuilds
+ // id 1 field by field (SparkRecordMergingUtils.mergePartialRecords)
with v taken from the
+ // base-file record, which carries the projection struct on the pushed
arm.
+ withSQLConf("hoodie.spark.sql.merge.into.partial.updates" -> "true") {
+ spark.sql(s"merge into $tableName t using (select 1 as id, 1003L as
ts) s on t.id = s.id " +
+ "when matched then update set ts = s.ts")
+ }
+ checkAnswer(s"select id, variant_get(v, '$$.k', 'string'), ts from
$tableName where id = 1")(
+ Seq(1, "x1", 1003L))
+ checkAnswer(s"select id, ${widePaths("v")} from $tableName where id =
1")(wideRow(1))
+ }
}
}
@@ -184,4 +213,78 @@ class TestVariantPushVariantIntoScan extends
HoodieSparkSqlTestBase {
assertPushed(s"select id, variant_get(s.inner, '$$.k', 'string') from
$tableName", pushed, tableName)
}
}
+
+ test("bootstrapped tables read the projection through the skeleton and data
file join") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // The record type is not swept: a METADATA_ONLY bootstrap on the SPARK
record type fails in the
+ // skeleton write (HUDI-5807), so every leg runs on the default record
type.
+ Seq("cow", "mor").foreach { tableType =>
+ Seq("false", "true").foreach { pushIntoScan =>
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) {
+ withTempDir { tmp =>
+ val tableName = generateTableName
+ val leg = s"bootstrap $tableType
pushVariantIntoScan=$pushIntoScan, $tableName"
+ val srcPath = s"${tmp.getCanonicalPath}/source"
+ val tablePath = s"${tmp.getCanonicalPath}/hudi"
+ spark.sql("""select cast(id as int) as id,
parse_json(concat('{"a":', id, '}')) as v, """ +
+ "1000L as ts from range(0, 10, 1, 1)")
+ .write.parquet(srcPath)
+ val writeOpts = Map(
+ DataSourceWriteOptions.TABLE_TYPE.key ->
+ (if (tableType == "mor")
DataSourceWriteOptions.MOR_TABLE_TYPE_OPT_VAL
+ else DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL),
+ HoodieWriteConfig.TBL_NAME.key -> tableName,
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "id",
+ DataSourceWriteOptions.ORDERING_FIELDS.key -> "ts",
+ DataSourceWriteOptions.KEYGENERATOR_CLASS_NAME.key ->
classOf[NonpartitionedKeyGenerator].getName,
+ // col stats is not supported with bootstrap operation
+ HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key ->
"false")
+ spark.emptyDataFrame.write.format("hudi")
+ .options(writeOpts)
+ .option(DataSourceWriteOptions.OPERATION.key,
DataSourceWriteOptions.BOOTSTRAP_OPERATION_OPT_VAL)
+ .option(HoodieBootstrapConfig.BASE_PATH.key, srcPath)
+ .mode(SaveMode.Overwrite)
+ .save(tablePath)
+ assert(createMetaClient(spark,
tablePath).getTableConfig.getBootstrapBasePath.isPresent,
+ s"[$leg] table should be bootstrapped from $srcPath")
+ // COW stays read-only: an upsert would rewrite the file group
into a regular base file and
+ // drop the skeleton and data file join. On MOR the upsert lands
in a log file, so the merged
+ // read joins skeleton, source file and log.
+ if (tableType == "mor") {
+ spark.sql("""select cast(id as int) as id, case when id = 9 then
null """ +
+ """else parse_json(concat('{"a":', 100 + id, '}')) end as v,
""" +
+ "1001L as ts from range(5, 10, 1, 1)")
+ .write.format("hudi")
+ .options(writeOpts)
+ .option(DataSourceWriteOptions.OPERATION.key,
DataSourceWriteOptions.UPSERT_OPERATION_OPT_VAL)
+ .mode(SaveMode.Append)
+ .save(tablePath)
+ }
+ val view = s"${tableName}_view"
+
spark.read.format("hudi").load(tablePath).createOrReplaceTempView(view)
+
+ val projectedSql = s"select id, variant_get(v, '$$.a', 'bigint')
from $view order by id"
+ if (tableType == "mor") {
+ checkAnswer(projectedSql)(
+ (0 until 9).map(id => Seq(id, if (id < 5) id.toLong else 100L
+ id)) :+ Seq(9, null): _*)
+ } else {
+ checkAnswer(projectedSql)((0 until 10).map(id => Seq(id,
id.toLong)): _*)
+ }
+ checkAnswer(s"select _hoodie_record_key, variant_get(v, '$$.a',
'bigint') from $view " +
+ "where id in (2, 7) order by id")(
+ Seq("2", 2L), Seq("7", if (tableType == "mor") 107L else 7L))
+ checkAnswer(s"select count(*) from $view where v is null")(
+ Seq(if (tableType == "mor") 1L else 0L))
+ // A wide read over the join: seven of the eight pushed paths are
absent from the row.
+ val wide = ('a' to 'h').map(c => s"try_variant_get(v, '$$.$c',
'bigint')").mkString(", ")
+ checkAnswer(s"select _hoodie_record_key, $wide from $view where id
= 2")(
+ Seq("2", 2L) ++ Seq.fill(7)(null))
+ assertPushed(projectedSql, pushIntoScan.toBoolean, leg)
+ spark.catalog.dropTempView(view)
+ }
+ }
+ }
+ }
+ }
}
