This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new d98f2f1b4752 refactor(spark): remove dead HoodieNestedSchemaPruning
rule (#19458)
d98f2f1b4752 is described below
commit d98f2f1b4752845167527a69f1912d6ad4980921
Author: voonhous <[email protected]>
AuthorDate: Mon Aug 3 16:02:09 2026 +0800
refactor(spark): remove dead HoodieNestedSchemaPruning rule (#19458)
* refactor(spark): remove dead HoodieNestedSchemaPruning rule
The rule only rewrites LogicalRelation(relation: HoodieBaseRelation) when
relation.canPruneRelationSchema is true. Since #14061 removed the
FILE_GROUP_READER_ENABLED fallback from DefaultSource and #17457 finished
the FileGroupReader migration, every non-metadata read returns a
HadoopFsRelation; the only remaining HoodieBaseRelation producers are
metadata-table reads and the schema-on-read branch, both of which
canPruneRelationSchema rejects, so the rule can never fire. Nested schema
pruning for users is performed by Spark's built-in SchemaPruning, since
the file-group-reader file format extends ParquetFileFormat.
Also remove the members stranded by the rule: canPruneRelationSchema,
updatePrunedDataSchema, dataSchema, type Relation and the prunedDataSchema
constructor parameter on HoodieBaseRelation and its subclasses,
MergeOnReadSnapshotRelation.isProjectionCompatible, and
BaseHoodieCatalystPlanUtils.projectOverSchema.
TestNestedSchemaPruningOptimization is kept unchanged; it covers the
Spark-native pruning path (FileSourceScanExec.requiredSchema) that serves
users today.
Fixes #19447
* refactor(spark): prune stranded members and harden nested-pruning coverage
Follow-up to the HoodieNestedSchemaPruning removal, addressing self-review
findings:
- Remove HoodieTableState.recordPayloadClassName: its only reader was the
deleted MergeOnReadSnapshotRelation.isProjectionCompatible. Drop the
now-unused ParquetFileFormat import in HoodieBaseRelation.
- SchemaHandlerTestBase#testMor: add a nested-narrowed projection block
("fare" pruned to its "amount" leaf) asserting that a
projection-incompatible custom merger re-expands to the full data
schema. The #7528 (HUDI-5443) payload gate retires with this PR because
FileGroupReaderSchemaHandler#generateRequiredSchema enforces the same
invariant through HoodieRecordMerger#isProjectionCompatible; this pins
it so a future HoodieBaseRelation-targeting pruning rule cannot
silently lose it.
- TestNestedSchemaPruningOptimization: replace the payload-invariant
DefaultHoodieRecordPayload case with a projection-incompatible custom
payload that merges through a log file end to end; note on the suite
that the pruning under test is Spark's own SchemaPruning rule.
* test(common): skip the nested-narrowed assertion for incompatible custom
mergers
The projection-incompatible custom-merger branch of generateRequiredSchema
returns the full data schema before looking at the requested schema, so the
nested-narrowed request asserted exactly what the flat-projection block
above
already pins. Run the nested-narrowed block only for the arms where the
narrowing has to survive.
Addresses review feedback on #19458.
* test(spark): describe the second checkAnswer as its own pruned query
The second checkAnswer issues an independent query that Spark prunes to
(id, item.price, ts), so it never observes leaves the first query's read
schema dropped; say what it actually pins.
Addresses review feedback on #19458.
* test(common): pin nested mandatory-field merge into a narrowed record
A merger declaring the nested leaf fare.currency mandatory against a request
narrowed to fare.amount forces generateRequiredSchema through the
appendFieldsToSchemaDedupNested/mergeSchemas collision path, which the
existing mandatory fields (all top-level, disjoint from fare) never reach.
Suggested by review on #19458.
---
.../common/table/read/SchemaHandlerTestBase.java | 45 ++++-
.../read/TestFileGroupReaderSchemaHandler.java | 14 ++
.../org/apache/hudi/BaseFileOnlyRelation.scala | 9 +-
.../scala/org/apache/hudi/HoodieBaseRelation.scala | 62 +------
.../hudi/HoodieHadoopFsRelationFactory.scala | 4 +-
.../hudi/MergeOnReadIncrementalRelationV1.scala | 10 +-
.../hudi/MergeOnReadIncrementalRelationV2.scala | 8 +-
.../apache/hudi/MergeOnReadSnapshotRelation.scala | 40 +---
.../spark/sql/BaseHoodieCatalystPlanUtils.scala | 9 +-
.../datasources/HoodieNestedSchemaPruning.scala | 206 ---------------------
.../spark/sql/hudi/analysis/HoodieAnalysis.scala | 5 +-
.../TestNestedSchemaPruningOptimization.scala | 24 ++-
12 files changed, 95 insertions(+), 341 deletions(-)
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/SchemaHandlerTestBase.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/SchemaHandlerTestBase.java
index f29c0b53a70b..047c08cdd873 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/SchemaHandlerTestBase.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/SchemaHandlerTestBase.java
@@ -130,6 +130,27 @@ public abstract class SchemaHandlerTestBase {
}
assertEquals(expectedRequiredSchema, schemaHandler.getRequiredSchema());
assertFalse(readerContext.getNeedsBootstrapMerge());
+
+ //read subset of columns with a nested-narrowed field, the shape Spark's
nested schema pruning
+ //requests: "fare" keeps only its "amount" leaf. Skipped for a
projection-incompatible custom
+ //merger: that branch returns the full data schema before looking at the
requested schema, so
+ //the flat-projection assertion above already covers it
+ if (!(mergeMode == CUSTOM && !isProjectionCompatible)) {
+ requestedSchema =
narrowFareToAmountOnly(generateProjectionSchema("begin_lat", "fare", "rider"));
+ schemaHandler = createSchemaHandler(readerContext, dataSchema,
requestedSchema, supportsParquetRowIndex);
+ if (mergeMode == EVENT_TIME_ORDERING && hasPrecombine) {
+ expectedRequiredSchema =
narrowFareToAmountOnly(generateProjectionSchema(hasBuiltInDelete, "begin_lat",
"fare", "rider", "_hoodie_record_key", "timestamp"));
+ } else if (mergeMode == EVENT_TIME_ORDERING || mergeMode ==
COMMIT_TIME_ORDERING) {
+ expectedRequiredSchema =
narrowFareToAmountOnly(generateProjectionSchema(hasBuiltInDelete, "begin_lat",
"fare", "rider", "_hoodie_record_key"));
+ } else {
+ expectedRequiredSchema =
narrowFareToAmountOnly(generateProjectionSchema("begin_lat", "fare", "rider",
"begin_lon", "_hoodie_record_key", "timestamp"));
+ }
+ if (supportsParquetRowIndex && mergeUseRecordPosition) {
+ expectedRequiredSchema = addPositionalMergeCol(expectedRequiredSchema);
+ }
+ assertEquals(expectedRequiredSchema, schemaHandler.getRequiredSchema());
+ assertFalse(readerContext.getNeedsBootstrapMerge());
+ }
}
public void testMorBootstrap(RecordMergeMode mergeMode,
@@ -255,7 +276,7 @@ public abstract class SchemaHandlerTestBase {
assertEquals(expectedBootstrapFields.getRight(),
bootstrapFields.getRight());
}
- private static void setupMORTable(RecordMergeMode mergeMode, boolean
hasPrecombine, HoodieTableConfig hoodieTableConfig) {
+ static void setupMORTable(RecordMergeMode mergeMode, boolean hasPrecombine,
HoodieTableConfig hoodieTableConfig) {
when(hoodieTableConfig.populateMetaFields()).thenReturn(true);
when(hoodieTableConfig.getRecordMergeMode()).thenReturn(mergeMode);
when(hoodieTableConfig.getTableVersion()).thenReturn(HoodieTableVersion.current());
@@ -271,7 +292,7 @@ public abstract class SchemaHandlerTestBase {
}
}
- private static HoodieRecordMerger mockRecordMerger(boolean
isProjectionCompatible, String[] mandatoryFields) throws IOException {
+ static HoodieRecordMerger mockRecordMerger(boolean isProjectionCompatible,
String[] mandatoryFields) throws IOException {
HoodieRecordMerger merger = mock(HoodieRecordMerger.class);
when(merger.isProjectionCompatible()).thenReturn(isProjectionCompatible);
when(merger.merge(any(), any(), any(), any())).thenReturn(null);
@@ -305,6 +326,26 @@ public abstract class SchemaHandlerTestBase {
return HoodieSchemaUtils.generateProjectionSchema(DATA_SCHEMA, fieldList);
}
+ /**
+ * Rebuilds the given projection schema, narrowing its "fare" record down to
the "amount" leaf to
+ * mimic the schema shape Spark's nested schema pruning requests.
+ */
+ static HoodieSchema narrowFareToAmountOnly(HoodieSchema projectionSchema) {
+ List<HoodieSchemaField> fields = new
ArrayList<>(projectionSchema.getFields().size());
+ for (HoodieSchemaField field : projectionSchema.getFields()) {
+ if (field.name().equals("fare")) {
+ HoodieSchema fare = field.schema();
+ HoodieSchema narrowedFare = HoodieSchema.createRecord(fare.getName(),
fare.getNamespace().orElse(null), fare.getDoc().orElse(null),
+
Collections.singletonList(HoodieSchemaUtils.createNewSchemaField(fare.getField("amount").get())));
+ fields.add(HoodieSchemaUtils.createNewSchemaField(field.name(),
narrowedFare, field.doc().orElse(null), field.defaultVal().orElse(null)));
+ } else {
+ fields.add(HoodieSchemaUtils.createNewSchemaField(field));
+ }
+ }
+ return HoodieSchema.createRecord(projectionSchema.getName(),
projectionSchema.getNamespace().orElse(null),
+ projectionSchema.getDoc().orElse(null), fields);
+ }
+
HoodieSchemaField getField(String fieldName) {
return DATA_SCHEMA.getField(fieldName).get();
}
diff --git
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
index 8a2549e16c14..8e9026cd10f7 100644
---
a/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
+++
b/hudi-common/src/test/java/org/apache/hudi/common/table/read/TestFileGroupReaderSchemaHandler.java
@@ -145,6 +145,20 @@ public class TestFileGroupReaderSchemaHandler extends
SchemaHandlerTestBase {
super.testMor(mergeMode, hasPrecombine, isProjectionCompatible,
mergeUseRecordPosition, supportsParquetRowIndex, hasBuiltInDelete);
}
+ @Test
+ public void testMorNestedMandatoryFieldMergesIntoNarrowedRecord() throws
IOException {
+ setupMORTable(RecordMergeMode.CUSTOM, false, hoodieTableConfig);
+ HoodieRecordMerger merger = mockRecordMerger(true, new String[]
{"fare.currency"});
+ HoodieReaderContext<String> readerContext =
createReaderContext(hoodieTableConfig, false, true, false, false, merger);
+
+ //the request narrows "fare" to its "amount" leaf while the merger
declares the sibling leaf
+ //"fare.currency" mandatory, so the handler has to merge the two
projections of "fare"
+ HoodieSchema requestedSchema =
narrowFareToAmountOnly(generateProjectionSchema("begin_lat", "fare", "rider"));
+ FileGroupReaderSchemaHandler schemaHandler =
createSchemaHandler(readerContext, DATA_SCHEMA, requestedSchema, false);
+ assertEquals(generateProjectionSchema("begin_lat", "fare", "rider"),
schemaHandler.getRequiredSchema());
+ assertFalse(readerContext.getNeedsBootstrapMerge());
+ }
+
@ParameterizedTest
@MethodSource("testMorParams")
public void testMorBootstrap(RecordMergeMode mergeMode,
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/BaseFileOnlyRelation.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/BaseFileOnlyRelation.scala
index 301f9c904fed..0a3c5d2b979e 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/BaseFileOnlyRelation.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/BaseFileOnlyRelation.scala
@@ -50,15 +50,13 @@ import org.apache.spark.sql.types.StructType
case class BaseFileOnlyRelation(override val sqlContext: SQLContext,
override val metaClient: HoodieTableMetaClient,
override val optParams: Map[String, String],
- private val userSchema: Option[StructType],
- private val prunedDataSchema:
Option[StructType] = None)
- extends HoodieBaseRelation(sqlContext, metaClient, optParams, userSchema,
prunedDataSchema)
+ private val userSchema: Option[StructType])
+ extends HoodieBaseRelation(sqlContext, metaClient, optParams, userSchema)
with SparkAdapterSupport {
case class HoodieBaseFileSplit(filePartition: FilePartition) extends
HoodieFileSplit
override type FileSplit = HoodieBaseFileSplit
- override type Relation = BaseFileOnlyRelation
// TODO(HUDI-3204) this is to override behavior (exclusively) for COW tables
to always extract
// partition values from partition path
@@ -74,9 +72,6 @@ case class BaseFileOnlyRelation(override val sqlContext:
SQLContext,
// Since Spark 3.4.0: FileIndexOptions.BASE_PATH_PARAM
val BASE_PATH_PARAM = "basePath"
- override def updatePrunedDataSchema(prunedSchema: StructType): Relation =
- this.copy(prunedDataSchema = Some(prunedSchema))
-
protected override def composeRDD(fileSplits: Seq[HoodieBaseFileSplit],
tableSchema: HoodieTableSchema,
requiredSchema: HoodieTableSchema,
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
index d6f045338350..e1e01103bdf2 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieBaseRelation.scala
@@ -45,7 +45,6 @@ import org.apache.hudi.exception.HoodieException
import org.apache.hudi.hadoop.fs.HadoopFSUtils
import org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePath
import org.apache.hudi.io.storage.HoodieSparkIOFactory
-import org.apache.hudi.metadata.HoodieTableMetadata
import org.apache.hudi.storage.{StoragePath, StoragePathInfo}
import org.apache.avro.generic.GenericRecord
@@ -64,7 +63,7 @@ import org.apache.spark.sql.catalyst.expressions.{Expression,
SubqueryExpression
import org.apache.spark.sql.execution.FileRelation
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.datasources.orc.OrcFileFormat
-import
org.apache.spark.sql.execution.datasources.parquet.{LegacyHoodieParquetFileFormat,
ParquetFileFormat}
+import
org.apache.spark.sql.execution.datasources.parquet.LegacyHoodieParquetFileFormat
import org.apache.spark.sql.hudi.{HoodieSqlCommonUtils, ProvidesHoodieConfig}
import org.apache.spark.sql.sources.{BaseRelation, Filter, PrunedFilteredScan}
import org.apache.spark.sql.types.StructType
@@ -81,7 +80,6 @@ case class HoodieTableState(tablePath: String,
recordKeyField: String,
orderingFields: List[String],
usesVirtualKeys: Boolean,
- recordPayloadClassName: String,
metadataConfig: HoodieMetadataConfig,
recordMergeImplClasses: List[String],
recordMergeStrategyId: String)
@@ -93,15 +91,13 @@ case class HoodieTableState(tablePath: String,
abstract class HoodieBaseRelation(val sqlContext: SQLContext,
val metaClient: HoodieTableMetaClient,
val optParams: Map[String, String],
- private val schemaSpec: Option[StructType],
- private val prunedDataSchema:
Option[StructType])
+ private val schemaSpec: Option[StructType])
extends BaseRelation
with FileRelation
with PrunedFilteredScan
with Logging {
type FileSplit <: HoodieFileSplit
- type Relation <: HoodieBaseRelation
imbueConfigs(sqlContext)
@@ -235,10 +231,6 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
shouldOmitPartitionColumns || shouldExtractPartitionValueFromPath ||
shouldUseBootstrapFastRead
}
- /**
- * NOTE: This fields are accessed by [[NestedSchemaPruning]] component which
is only enabled for
- * Spark >= 3.1
- */
protected lazy val (fileFormat: FileFormat, fileFormatClassName: String) =
metaClient.getTableConfig.getBaseFileFormat match {
case HoodieFileFormat.ORC => (new OrcFileFormat, "orc")
@@ -268,7 +260,6 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
recordKeyField = recordKeyField,
orderingFields = orderingFields,
usesVirtualKeys = !tableConfig.populateMetaFields(),
- recordPayloadClassName = tableConfig.getPayloadClass,
metadataConfig = fileIndex.getMetadataConfig,
recordMergeImplClasses = recordMergerImpls,
recordMergeStrategyId = tableConfig.getRecordMergeStrategyId
@@ -297,39 +288,9 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
*/
def hasSchemaOnRead: Boolean = internalSchemaOpt.isDefined
- /**
- * Data schema is determined as the actual schema of the Table's Data Files
(for ex, parquet/orc/etc);
- *
- * In cases when partition values are not persisted w/in the data files,
data-schema is defined as
- * <pre>table's schema - partition columns</pre>
- *
- * Check scala-doc for [[shouldExtractPartitionValuesFromPartitionPath]] for
more details
- */
- def dataSchema: StructType = if
(shouldExtractPartitionValuesFromPartitionPath) {
- prunePartitionColumns(tableStructSchema)
- } else {
- tableStructSchema
- }
-
- /**
- * Determines whether relation's schema could be pruned by Spark's Optimizer
- */
- def canPruneRelationSchema: Boolean =
- !HoodieTableMetadata.isMetadataTable(basePath.toString) &&
- (fileFormat.isInstanceOf[ParquetFileFormat] ||
fileFormat.isInstanceOf[OrcFileFormat]) &&
- // NOTE: In case this relation has already been pruned there's no point
in pruning it again
- prunedDataSchema.isEmpty &&
- // TODO(HUDI-5421) internal schema doesn't support nested schema pruning
currently
- !hasSchemaOnRead
-
override def sizeInBytes: Long = fileIndex.sizeInBytes
- override def schema: StructType = {
- // NOTE: Optimizer could prune the schema (applying for ex,
[[NestedSchemaPruning]] rule) setting new updated
- // schema in-place (via [[setPrunedDataSchema]] method), therefore
we have to make sure that we pick
- // pruned data schema (if present) over the standard table's one
- prunedDataSchema.getOrElse(tableStructSchema)
- }
+ override def schema: StructType = tableStructSchema
/**
* This method controls whether relation will be producing
@@ -355,13 +316,8 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
// (!) Please note, however, that it's critical to avoid _reordering_ of
the requested columns as this
// will break the upstream projection
val targetColumns: Array[String] = appendMandatoryColumns(requiredColumns)
- // NOTE: We explicitly fallback to default table's Avro schema to make
sure we avoid unnecessary Catalyst > Avro
- // schema conversion, which is lossy in nature (for ex, it doesn't
preserve original Avro type-names) and
- // could have an effect on subsequent de-/serializing records in
some exotic scenarios (when Avro unions
- // w/ more than 2 types are involved)
- val sourceSchema = prunedDataSchema.map(s => convertToHoodieSchema(s,
tableName)).getOrElse(tableSchema)
val (requiredProjectedSchema, requiredStructSchema,
requiredInternalSchema) =
- projectSchema(Either.cond(internalSchemaOpt.isDefined,
internalSchemaOpt.get, sourceSchema), targetColumns)
+ projectSchema(Either.cond(internalSchemaOpt.isDefined,
internalSchemaOpt.get, tableSchema), targetColumns)
val filterExpressions = convertToExpressions(filters)
val (partitionFilters, dataFilters) =
filterExpressions.partition(isPartitionPredicate)
@@ -509,16 +465,6 @@ abstract class HoodieBaseRelation(val sqlContext:
SQLContext,
}
}
- /**
- * Hook for Spark's Optimizer to update expected relation schema after
pruning
- *
- * NOTE: Only limited number of optimizations in respect to schema pruning
could be performed
- * internally w/in the relation itself w/o consideration for how the
relation output is used.
- * Therefore more advanced optimizations (like
[[NestedSchemaPruning]]) have to be carried out
- * by Spark's Optimizer holistically evaluating Spark's [[LogicalPlan]]
- */
- def updatePrunedDataSchema(prunedSchema: StructType): Relation
-
protected def createBaseFileReaders(tableSchema: HoodieTableSchema,
requiredSchema: HoodieTableSchema,
requestedColumns: Array[String],
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
index c35d57bb101a..7e475dc9971c 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieHadoopFsRelationFactory.scala
@@ -335,7 +335,7 @@ class
HoodieMergeOnReadIncrementalHadoopFsRelationFactoryV2(override val sqlCont
isBootstrap:
Boolean,
rangeType:
RangeType = RangeType.OPEN_CLOSED)
extends HoodieMergeOnReadIncrementalHadoopFsRelationFactory(sqlContext,
metaClient, options, schemaSpec, isBootstrap,
- MergeOnReadIncrementalRelationV2(sqlContext, options, metaClient,
schemaSpec, None, rangeType))
+ MergeOnReadIncrementalRelationV2(sqlContext, options, metaClient,
schemaSpec, rangeType))
class HoodieMergeOnReadCDCHadoopFsRelationFactory(override val sqlContext:
SQLContext,
override val metaClient:
HoodieTableMetaClient,
@@ -435,7 +435,7 @@ class
HoodieCopyOnWriteIncrementalHadoopFsRelationFactoryV2(override val sqlCont
isBootstrap:
Boolean,
rangeType:
RangeType = RangeType.OPEN_CLOSED)
extends HoodieCopyOnWriteIncrementalHadoopFsRelationFactory(sqlContext,
metaClient, options, schemaSpec, isBootstrap,
- MergeOnReadIncrementalRelationV2(sqlContext, options, metaClient,
schemaSpec, None, rangeType))
+ MergeOnReadIncrementalRelationV2(sqlContext, options, metaClient,
schemaSpec, rangeType))
class HoodieCopyOnWriteCDCHadoopFsRelationFactory(override val sqlContext:
SQLContext,
override val metaClient:
HoodieTableMetaClient,
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
index e0fc7ceb6824..fabdfdadfe3c 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV1.scala
@@ -48,16 +48,10 @@ import scala.collection.immutable
case class MergeOnReadIncrementalRelationV1(override val sqlContext:
SQLContext,
override val optParams:
Map[String, String],
override val metaClient:
HoodieTableMetaClient,
- private val userSchema:
Option[StructType],
- private val prunedDataSchema:
Option[StructType] = None)
- extends BaseMergeOnReadSnapshotRelation(sqlContext, optParams, metaClient,
userSchema, prunedDataSchema)
+ private val userSchema:
Option[StructType])
+ extends BaseMergeOnReadSnapshotRelation(sqlContext, optParams, metaClient,
userSchema)
with HoodieIncrementalRelationV1Trait with MergeOnReadIncrementalRelation {
- override type Relation = MergeOnReadIncrementalRelationV1
-
- override def updatePrunedDataSchema(prunedSchema: StructType): Relation =
- this.copy(prunedDataSchema = Some(prunedSchema))
-
override protected def timeline: HoodieTimeline = {
if (fullTableScan) {
handleHollowCommitIfNeeded(metaClient.getCommitsAndCompactionTimeline,
metaClient, hollowCommitHandling)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
index aea594d9157c..b465e38623a1 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadIncrementalRelationV2.scala
@@ -51,16 +51,10 @@ case class MergeOnReadIncrementalRelationV2(override val
sqlContext: SQLContext,
override val optParams:
Map[String, String],
override val metaClient:
HoodieTableMetaClient,
private val userSchema:
Option[StructType],
- private val prunedDataSchema:
Option[StructType] = None,
override val rangeType: RangeType
= RangeType.OPEN_CLOSED)
- extends BaseMergeOnReadSnapshotRelation(sqlContext, optParams, metaClient,
userSchema, prunedDataSchema)
+ extends BaseMergeOnReadSnapshotRelation(sqlContext, optParams, metaClient,
userSchema)
with HoodieIncrementalRelationV2Trait with MergeOnReadIncrementalRelation {
- override type Relation = MergeOnReadIncrementalRelationV2
-
- override def updatePrunedDataSchema(prunedSchema: StructType): Relation =
- this.copy(prunedDataSchema = Some(prunedSchema))
-
override protected def timeline: HoodieTimeline = {
if (fullTableScan) {
metaClient.getCommitsAndCompactionTimeline
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala
index 66ca8c8d6a25..aa96512ca18f 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/MergeOnReadSnapshotRelation.scala
@@ -19,8 +19,8 @@
package org.apache.hudi
import org.apache.hudi.HoodieConversionUtils.toScalaOption
-import org.apache.hudi.MergeOnReadSnapshotRelation.{createPartitionedFile,
isProjectionCompatible}
-import org.apache.hudi.common.model.{FileSlice, HoodieLogFile,
OverwriteWithLatestAvroPayload}
+import org.apache.hudi.MergeOnReadSnapshotRelation.createPartitionedFile
+import org.apache.hudi.common.model.{FileSlice, HoodieLogFile}
import org.apache.hudi.common.table.HoodieTableMetaClient
import org.apache.hudi.storage.StoragePath
@@ -42,14 +42,8 @@ case class HoodieMergeOnReadFileSplit(dataFile:
Option[PartitionedFile],
case class MergeOnReadSnapshotRelation(override val sqlContext: SQLContext,
override val optParams: Map[String,
String],
override val metaClient:
HoodieTableMetaClient,
- private val userSchema:
Option[StructType],
- private val prunedDataSchema:
Option[StructType] = None)
- extends BaseMergeOnReadSnapshotRelation(sqlContext, optParams, metaClient,
userSchema, prunedDataSchema) {
-
- override type Relation = MergeOnReadSnapshotRelation
-
- override def updatePrunedDataSchema(prunedSchema: StructType): Relation =
- this.copy(prunedDataSchema = Some(prunedSchema))
+ private val userSchema:
Option[StructType])
+ extends BaseMergeOnReadSnapshotRelation(sqlContext, optParams, metaClient,
userSchema) {
override protected def shouldIncludeLogFiles(): Boolean = {
true
@@ -67,9 +61,8 @@ case class MergeOnReadSnapshotRelation(override val
sqlContext: SQLContext,
abstract class BaseMergeOnReadSnapshotRelation(sqlContext: SQLContext,
optParams: Map[String, String],
metaClient:
HoodieTableMetaClient,
- userSchema: Option[StructType],
- prunedDataSchema:
Option[StructType])
- extends HoodieBaseRelation(sqlContext, metaClient, optParams, userSchema,
prunedDataSchema) {
+ userSchema: Option[StructType])
+ extends HoodieBaseRelation(sqlContext, metaClient, optParams, userSchema) {
override type FileSplit = HoodieMergeOnReadFileSplit
@@ -97,12 +90,6 @@ abstract class BaseMergeOnReadSnapshotRelation(sqlContext:
SQLContext,
protected val mergeType: String =
optParams.getOrElse(DataSourceReadOptions.REALTIME_MERGE.key,
DataSourceReadOptions.REALTIME_MERGE.defaultValue)
- /**
- * Determines whether relation's schema could be pruned by Spark's Optimizer
- */
- override def canPruneRelationSchema: Boolean =
- super.canPruneRelationSchema && isProjectionCompatible(tableState)
-
protected override def composeRDD(fileSplits:
Seq[HoodieMergeOnReadFileSplit],
tableSchema: HoodieTableSchema,
requiredSchema: HoodieTableSchema,
@@ -152,21 +139,6 @@ abstract class BaseMergeOnReadSnapshotRelation(sqlContext:
SQLContext,
object MergeOnReadSnapshotRelation extends SparkAdapterSupport {
- /**
- * List of [[HoodieRecordPayload]] classes capable of merging projected
records:
- * in some cases, when for example, user is only interested in a handful of
columns rather
- * than the full row we will be able to optimize data throughput by only
fetching the required
- * columns. However, to properly fulfil MOR semantic particular
[[HoodieRecordPayload]] in
- * question should be able to merge records based on just such projected
representation (including
- * columns required for merging, such as primary-key, pre-combine key, etc)
- */
- private val projectionCompatiblePayloadClasses: Set[String] = Seq(
- classOf[OverwriteWithLatestAvroPayload]
- ).map(_.getName).toSet
-
- def isProjectionCompatible(tableState: HoodieTableState): Boolean =
-
projectionCompatiblePayloadClasses.contains(tableState.recordPayloadClassName)
-
def createPartitionedFile(partitionValues: InternalRow,
filePath: StoragePath,
start: Long,
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystPlanUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystPlanUtils.scala
index c8ceb713a6a3..6b8644519d2a 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystPlanUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/BaseHoodieCatalystPlanUtils.scala
@@ -22,7 +22,7 @@ import org.apache.hudi.SparkAdapterSupport
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.{ResolvedTable,
TableOutputResolver}
import org.apache.spark.sql.catalyst.catalog.CatalogStorageFormat
-import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet,
Expression, ProjectionOverSchema}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
import org.apache.spark.sql.catalyst.plans.JoinType
import org.apache.spark.sql.catalyst.plans.logical.{CreateIndex, DropIndex,
HoodieShowIndexes, InsertIntoStatement, Join, JoinHint, LogicalPlan,
RefreshIndex}
import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog}
@@ -30,16 +30,9 @@ import org.apache.spark.sql.execution.{ExtendedMode,
SimpleMode}
import org.apache.spark.sql.execution.command.{CreateTableLikeCommand,
ExplainCommand, RepairTableCommand}
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructType
trait BaseHoodieCatalystPlanUtils extends HoodieCatalystPlansUtils {
- /**
- * Instantiates [[ProjectionOverSchema]] utility
- */
- def projectOverSchema(schema: StructType, output: AttributeSet):
ProjectionOverSchema =
- ProjectionOverSchema(schema, output)
-
/**
* Un-applies [[ResolvedTable]] that had its signature changed in Spark 3.2
*/
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieNestedSchemaPruning.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieNestedSchemaPruning.scala
deleted file mode 100644
index 42e59eea5c51..000000000000
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/HoodieNestedSchemaPruning.scala
+++ /dev/null
@@ -1,206 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.execution.datasources
-
-import org.apache.hudi.{HoodieBaseRelation, SparkAdapterSupport}
-
-import org.apache.spark.sql.BaseHoodieCatalystPlanUtils
-import org.apache.spark.sql.catalyst.expressions.{And, AttributeReference,
AttributeSet, Expression, NamedExpression, ProjectionOverSchema}
-import org.apache.spark.sql.catalyst.planning.PhysicalOperation
-import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan,
Project}
-import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.sources.BaseRelation
-import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
-import org.apache.spark.sql.util.SchemaUtils.restoreOriginalOutputNames
-
-/**
- * Prunes unnecessary physical columns given a [[PhysicalOperation]] over a
data source relation.
- * By "physical column", we mean a column as defined in the data source format
like Parquet format
- * or ORC format. For example, in Spark SQL, a root-level Parquet column
corresponds to a SQL
- * column, and a nested Parquet column corresponds to a [[StructField]].
- *
- * NOTE: This class is borrowed from Spark 3.2.1, with modifications adapting
it to handle [[HoodieBaseRelation]],
- * instead of [[HadoopFsRelation]]
- */
-class HoodieNestedSchemaPruning extends Rule[LogicalPlan] {
- import org.apache.spark.sql.catalyst.expressions.SchemaPruning._
-
- override def apply(plan: LogicalPlan): LogicalPlan =
- if (conf.nestedSchemaPruningEnabled) {
- apply0(plan)
- } else {
- plan
- }
-
- private def apply0(plan: LogicalPlan): LogicalPlan =
- plan transformDown {
- // NOTE: The relation is matched by type rather than by destructuring
[[LogicalRelation]],
- // since the arity of its unapply differs across the Spark
versions this module
- // compiles against. This is modified to accommodate for Hudi's
custom relations,
- // given that original [[NestedSchemaPruning]] rule is tightly
coupled w/
- // [[HadoopFsRelation]]
- // TODO generalize to any file-based relation
- case op @ PhysicalOperation(projects, filters, l: LogicalRelation) =>
- l.relation match {
- case relation: HoodieBaseRelation if relation.canPruneRelationSchema
=>
- prunePhysicalColumns(l.output, projects, filters,
relation.dataSchema,
- prunedDataSchema => {
- val prunedRelation =
- relation.updatePrunedDataSchema(prunedSchema =
prunedDataSchema)
- buildPrunedRelation(l, prunedRelation)
- }).getOrElse(op)
- case _ => op
- }
- }
-
- // Prune the given output to make it consistent with `requiredSchema`.
- private def getPrunedOutput(output: Seq[AttributeReference],
- requiredSchema: StructType):
Seq[AttributeReference] = {
- // We need to replace the expression ids of the pruned relation output
attributes
- // with the expression ids of the original relation output attributes so
that
- // references to the original relation's output are not broken
- val outputIdMap = output.map(att => (att.name, att.exprId)).toMap
- // NOTE: The attributes are constructed inline (equivalent to
StructType#toAttributes before
- // Spark 3.5 and DataTypeUtils#toAttributes since, see SPARK-44353)
so that this code
- // compiles against every supported Spark version
- requiredSchema
- .map(f => AttributeReference(f.name, f.dataType, f.nullable,
f.metadata)())
- .map {
- case att if outputIdMap.contains(att.name) =>
- att.withExprId(outputIdMap(att.name))
- case att => att
- }
- }
-
- /**
- * This method returns optional logical plan. `None` is returned if no
nested field is required or
- * all nested fields are required.
- */
- private def prunePhysicalColumns(output: Seq[AttributeReference],
- projects: Seq[NamedExpression],
- filters: Seq[Expression],
- dataSchema: StructType,
- outputRelationBuilder: StructType =>
LogicalRelation): Option[LogicalPlan] = {
- val (normalizedProjects, normalizedFilters) =
- normalizeAttributeRefNames(output, projects, filters)
- val requestedRootFields = identifyRootFields(normalizedProjects,
normalizedFilters)
-
- // If requestedRootFields includes a nested field, continue. Otherwise,
- // return op
- if (requestedRootFields.exists { root: RootField => !root.derivedFromAtt
}) {
- val prunedDataSchema = pruneSchema(dataSchema, requestedRootFields)
-
- // If the data schema is different from the pruned data schema,
continue. Otherwise,
- // return op. We effect this comparison by counting the number of "leaf"
fields in
- // each schemata, assuming the fields in prunedDataSchema are a subset
of the fields
- // in dataSchema.
- if (countLeaves(dataSchema) > countLeaves(prunedDataSchema)) {
- val planUtils =
SparkAdapterSupport.sparkAdapter.getCatalystPlanUtils.asInstanceOf[BaseHoodieCatalystPlanUtils]
-
- val prunedRelation = outputRelationBuilder(prunedDataSchema)
- val projectionOverSchema =
planUtils.projectOverSchema(prunedDataSchema, AttributeSet(output))
-
- Some(buildNewProjection(projects, normalizedProjects,
normalizedFilters,
- prunedRelation, projectionOverSchema))
- } else {
- None
- }
- } else {
- None
- }
- }
-
- /**
- * Normalizes the names of the attribute references in the given projects
and filters to reflect
- * the names in the given logical relation. This makes it possible to
compare attributes and
- * fields by name. Returns a tuple with the normalized projects and filters,
respectively.
- */
- private def normalizeAttributeRefNames(output: Seq[AttributeReference],
- projects: Seq[NamedExpression],
- filters: Seq[Expression]):
(Seq[NamedExpression], Seq[Expression]) = {
- val normalizedAttNameMap = output.map(att => (att.exprId, att.name)).toMap
- val normalizedProjects = projects.map(_.transform {
- case att: AttributeReference if
normalizedAttNameMap.contains(att.exprId) =>
- att.withName(normalizedAttNameMap(att.exprId))
- }).map { case expr: NamedExpression => expr }
- val normalizedFilters = filters.map(_.transform {
- case att: AttributeReference if
normalizedAttNameMap.contains(att.exprId) =>
- att.withName(normalizedAttNameMap(att.exprId))
- })
- (normalizedProjects, normalizedFilters)
- }
-
- /**
- * Builds the new output [[Project]] Spark SQL operator that has the
`leafNode`.
- */
- private def buildNewProjection(projects: Seq[NamedExpression],
- normalizedProjects: Seq[NamedExpression],
- filters: Seq[Expression],
- prunedRelation: LogicalRelation,
- projectionOverSchema: ProjectionOverSchema):
Project = {
- // Construct a new target for our projection by rewriting and
- // including the original filters where available
- val projectionChild =
- if (filters.nonEmpty) {
- val projectedFilters = filters.map(_.transformDown {
- case projectionOverSchema(expr) => expr
- })
- val newFilterCondition = projectedFilters.reduce(And)
- Filter(newFilterCondition, prunedRelation)
- } else {
- prunedRelation
- }
-
- // Construct the new projections of our Project by
- // rewriting the original projections
- val newProjects = normalizedProjects.map(_.transformDown {
- case projectionOverSchema(expr) => expr
- }).map { case expr: NamedExpression => expr }
-
- if (log.isDebugEnabled) {
- logDebug(s"New
projects:\n${newProjects.map(_.treeString).mkString("\n")}")
- }
-
- Project(restoreOriginalOutputNames(newProjects, projects.map(_.name)),
projectionChild)
- }
-
- /**
- * Builds a pruned logical relation from the output of the output relation
and the schema of the
- * pruned base relation.
- */
- private def buildPrunedRelation(outputRelation: LogicalRelation,
- prunedBaseRelation: BaseRelation):
LogicalRelation = {
- val prunedOutput = getPrunedOutput(outputRelation.output,
prunedBaseRelation.schema)
- outputRelation.copy(relation = prunedBaseRelation, output = prunedOutput)
- }
-
- /**
- * Counts the "leaf" fields of the given dataType. Informally, this is the
- * number of fields of non-complex data type in the tree representation of
- * [[DataType]].
- */
- private def countLeaves(dataType: DataType): Int = {
- dataType match {
- case array: ArrayType => countLeaves(array.elementType)
- case map: MapType => countLeaves(map.keyType) +
countLeaves(map.valueType)
- case struct: StructType =>
- struct.map(field => countLeaves(field.dataType)).sum
- case _ => 1
- }
- }
-}
diff --git
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
index c6eb3c3122ee..f9d99cbc0602 100644
---
a/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
+++
b/hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/analysis/HoodieAnalysis.scala
@@ -30,7 +30,7 @@ import
org.apache.spark.sql.catalyst.optimizer.ReplaceExpressions
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.command._
-import org.apache.spark.sql.execution.datasources.{CreateTable,
HoodieNestedSchemaPruning, LogicalRelation}
+import org.apache.spark.sql.execution.datasources.{CreateTable,
LogicalRelation}
import org.apache.spark.sql.hudi.HoodieSqlCommonUtils.{isMetaField,
removeMetaFields}
import org.apache.spark.sql.hudi.analysis.HoodieAnalysis.{sparkAdapter,
MatchCreateIndex, MatchCreateTableLike, MatchDropIndex,
MatchInsertIntoStatement, MatchMergeIntoTable, MatchRefreshIndex,
MatchShowIndexes, ResolvesToHudiTable}
import org.apache.spark.sql.hudi.blob.ReadBlobRule
@@ -142,9 +142,6 @@ object HoodieAnalysis extends SparkAdapterSupport {
// Default rules
)
- val nestedSchemaPruningRule = new HoodieNestedSchemaPruning
- rules += (_ => nestedSchemaPruningRule)
-
// NOTE: [[HoodiePruneFileSourcePartitions]] is a replica in kind to
Spark's
// [[PruneFileSourcePartitions]] and as such should be executed at
the same stage.
// However, currently Spark doesn't allow [[SparkSessionExtensions]]
to inject into
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
index 7bdbef9cd008..be8cf2eda5a7 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/TestNestedSchemaPruningOptimization.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.hudi.common
+import org.apache.hudi.common.table.read.CustomPayloadForTesting
import org.apache.hudi.config.HoodieWriteConfig
import org.apache.spark.sql.DataFrame
@@ -25,6 +26,11 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType,
StringType, StructField, StructType}
import org.junit.jupiter.api.Assertions.assertEquals
+/**
+ * Verifies nested schema pruning on Hudi tables. The pruning itself is
performed by Spark's
+ * built-in SchemaPruning rule: the file-group-reader based file format
extends ParquetFileFormat,
+ * so Hudi does not ship a nested schema pruning rule of its own.
+ */
class TestNestedSchemaPruningOptimization extends HoodieSparkSqlTestBase {
// NOTE: We disable WCE once for the whole suite so the executed plans stay
a plain Project over the
@@ -58,26 +64,34 @@ class TestNestedSchemaPruningOptimization extends
HoodieSparkSqlTestBase {
}
}
- test("Test nested schema pruning with DefaultHoodieRecordPayload") {
+ test("Test nested schema pruning with a projection-incompatible custom
payload") {
withTempDir { tmp =>
val tableName = generateTableName
val tablePath = s"${tmp.getCanonicalPath}/$tableName"
- // NOTE: On the file-group-reader based read path the payload class does
not affect nested
- // schema pruning, so the read schema is pruned the same way as
with the default payload
+ // NOTE: A payload class outside the well-known set puts the table in
CUSTOM merge mode, whose
+ // merger is not projection compatible, so the file group reader
merges on the full
+ // table schema internally
(FileGroupReaderSchemaHandler#generateRequiredSchema) and
+ // projects the merged rows back down to the pruned read schema
afterwards
createTableWithNestedStructSchema("mor", tableName, tablePath,
- Map(HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key ->
"org.apache.hudi.common.model.DefaultHoodieRecordPayload"))
+ Map(HoodieWriteConfig.WRITE_PAYLOAD_CLASS_NAME.key ->
classOf[CustomPayloadForTesting].getName),
+ populateMetaFields = true)
+
+ // The update writes a log file, so the pruned reads below actually
merge through that gate
+ spark.sql(s"UPDATE $tableName SET ts = 123457 WHERE id = 1")
val selectDF = spark.sql(s"SELECT id, item.name FROM $tableName")
+ // Spark still prunes the scan schema; the full-schema requirement is
internal to the reader
val expectedSchema = StructType(Seq(
StructField("id", IntegerType, nullable = true),
StructField("item", StructType(Seq(StructField("name", StringType,
nullable = false))), nullable = true)
))
-
assertPrunedReadSchema(selectDF, tableName, expectedSchema)
checkAnswer(s"SELECT id, item.name FROM $tableName")(Seq(1, "a1"))
+ // A second query pruned to different leaves still observes the
correctly merged values
+ checkAnswer(s"SELECT id, item.price, ts FROM $tableName")(Seq(1, 10,
123457))
}
}
