[SPARK-39265][SQL] Support vectorized Parquet scans with DEFAULT values

[dtenedor]

What changes were proposed in this pull request?

Support vectorized Parquet scans when the table schema has associated DEFAULT column values.

Example:

create table t(i int) using parquet;
insert into t values(42);
alter table t add column s string default concat('abc', def');
select * from t;
> 42, 'abcdef'

Why are the changes needed?

This change makes it easier to build, query, and maintain tables backed by Parquet data.

Does this PR introduce any user-facing change?

Yes.

How was this patch tested?

This PR includes new test coverage.

[dtenedor]

Synced latest changes from master, this PR no longer depends on any other unmerged PRs anymore

[HyukjinKwon]

cc @sadikovi too FYI

[AmplabJenkins]

Can one of the admins verify this patch?

[sadikovi]

Also, can we add a test to check that the DEFAULT values work? Thanks.

[dtenedor]

Also, can we add a test to check that the DEFAULT values work? Thanks.

@sadikovi Sure, this is done in InsertSuite, by adding a new configuration to the case covering Parquet files (previously it only covered the non-vectorized case, but now with Config(None) it also runs the test over the vectorized case as well):

      TestCase(
        dataSource = "parquet",
        Seq(
          Config(
            None),
          Config(
            Some(SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key -> "false"),
            insertNullsToStorage = false)))

[sadikovi]

LGTM. Thanks for updating the test.

[HyukjinKwon]

The test log is a bit messy .. just copying and pasting the error I saw:

2022-06-02T02:22:55.9427269Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m- SPARK-38336 INSERT INTO statements with tables with default columns: negative tests *** FAILED *** (13 milliseconds)�[0m�[0m
2022-06-02T02:22:55.9442627Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  org.apache.spark.sql.AnalysisException: Failed to execute CREATE TABLE command because the destination table column s has a DEFAULT value of badvalue which fails to resolve as a valid expression: [MISSING_COLUMN] Column 'badvalue' does not exist. Did you mean one of the following? []; line 1 pos 0�[0m�[0m
2022-06-02T02:22:55.9443938Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.util.ResolveDefaultColumns$.analyze(ResolveDefaultColumnsUtil.scala:141)�[0m�[0m
2022-06-02T02:22:55.9445083Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.util.ResolveDefaultColumns$.$anonfun$constantFoldCurrentDefaultsToExistDefaults$1(ResolveDefaultColumnsUtil.scala:96)�[0m�[0m
2022-06-02T02:22:55.9449272Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:286)�[0m�[0m
2022-06-02T02:22:55.9451745Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.IndexedSeqOptimized.foreach(IndexedSeqOptimized.scala:36)�[0m�[0m
2022-06-02T02:22:55.9452379Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.IndexedSeqOptimized.foreach$(IndexedSeqOptimized.scala:33)�[0m�[0m
2022-06-02T02:22:55.9452963Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:198)�[0m�[0m
2022-06-02T02:22:55.9453505Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.TraversableLike.map(TraversableLike.scala:286)�[0m�[0m
2022-06-02T02:22:55.9454049Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.TraversableLike.map$(TraversableLike.scala:279)�[0m�[0m
2022-06-02T02:22:55.9454583Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:198)�[0m�[0m
2022-06-02T02:22:55.9494729Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.util.ResolveDefaultColumns$.constantFoldCurrentDefaultsToExistDefaults(ResolveDefaultColumnsUtil.scala:94)�[0m�[0m
2022-06-02T02:22:55.9495832Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.execution.datasources.DataSourceAnalysis$$anonfun$apply$1.applyOrElse(DataSourceStrategy.scala:153)�[0m�[0m
2022-06-02T02:22:55.9496639Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.execution.datasources.DataSourceAnalysis$$anonfun$apply$1.applyOrElse(DataSourceStrategy.scala:148)�[0m�[0m
2022-06-02T02:22:55.9497537Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.$anonfun$resolveOperatorsDownWithPruning$2(AnalysisHelper.scala:170)�[0m�[0m
2022-06-02T02:22:55.9498263Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:176)�[0m�[0m
2022-06-02T02:22:55.9499007Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.$anonfun$resolveOperatorsDownWithPruning$1(AnalysisHelper.scala:170)�[0m�[0m
2022-06-02T02:22:55.9499830Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper$.allowInvokingTransformsInAnalyzer(AnalysisHelper.scala:323)�[0m�[0m
2022-06-02T02:22:55.9500716Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.resolveOperatorsDownWithPruning(AnalysisHelper.scala:168)�[0m�[0m
2022-06-02T02:22:55.9501658Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.resolveOperatorsDownWithPruning$(AnalysisHelper.scala:164)�[0m�[0m
2022-06-02T02:22:55.9502569Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolveOperatorsDownWithPruning(LogicalPlan.scala:30)�[0m�[0m
2022-06-02T02:22:55.9503469Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.resolveOperatorsWithPruning(AnalysisHelper.scala:99)�[0m�[0m
2022-06-02T02:22:55.9504354Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.resolveOperatorsWithPruning$(AnalysisHelper.scala:96)�[0m�[0m
2022-06-02T02:22:55.9505333Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolveOperatorsWithPruning(LogicalPlan.scala:30)�[0m�[0m
2022-06-02T02:22:55.9506175Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.resolveOperators(AnalysisHelper.scala:76)�[0m�[0m
2022-06-02T02:22:55.9506968Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.resolveOperators$(AnalysisHelper.scala:75)�[0m�[0m
2022-06-02T02:22:55.9507748Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolveOperators(LogicalPlan.scala:30)�[0m�[0m
2022-06-02T02:22:55.9508709Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.execution.datasources.DataSourceAnalysis.apply(DataSourceStrategy.scala:148)�[0m�[0m
2022-06-02T02:22:55.9509477Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.execution.datasources.DataSourceAnalysis.apply(DataSourceStrategy.scala:64)�[0m�[0m
2022-06-02T02:22:55.9510186Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$2(RuleExecutor.scala:211)�[0m�[0m
2022-06-02T02:22:55.9510811Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.LinearSeqOptimized.foldLeft(LinearSeqOptimized.scala:126)�[0m�[0m
2022-06-02T02:22:55.9511420Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.LinearSeqOptimized.foldLeft$(LinearSeqOptimized.scala:122)�[0m�[0m
2022-06-02T02:22:55.9511976Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.immutable.List.foldLeft(List.scala:91)�[0m�[0m
2022-06-02T02:22:55.9512576Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1(RuleExecutor.scala:208)�[0m�[0m
2022-06-02T02:22:55.9513240Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at org.apache.spark.sql.catalyst.rules.RuleExecutor.$anonfun$execute$1$adapted(RuleExecutor.scala:200)�[0m�[0m
2022-06-02T02:22:55.9513896Z �[0m[�[0m�[0minfo�[0m] �[0m�[0m�[31m  at scala.collection.immutable.List.foreach(List.scala:431)�[0m�[0m

[dtenedor]

@HyukjinKwon the CI passes now :)

[dtenedor]

@gengliangwang @HyukjinKwon @cloud-fan can someone please merge this in (or leave more review comment(s) if desired for another pass)?

[gengliangwang]

Thanks for the work!