Spark dataframe count on entries metadata table throws an IllegalArgumentException for partitioned tables

[shardulm94]

I have created a test case to demonstrate this issue at shardulm94@6e0d0c2

The failing test can be run using ./gradlew :iceberg-spark2:test --tests="*TestIcebergSourceHiveTables24*"

> Task :iceberg-spark2:test

org.apache.iceberg.spark.source.TestIcebergSourceHiveTables24 > testCountEntriesPartitionedTable FAILED
    org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 38.0 failed 1 times, most recent failure: Lost task 0.0 in stage 38.0 (TID 447, localhost, executor driver): java.lang.IllegalArgumentException: Missing required field: data_file
        at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:217)
        at org.apache.iceberg.avro.BuildAvroProjection.record(BuildAvroProjection.java:98)
        at org.apache.iceberg.avro.BuildAvroProjection.record(BuildAvroProjection.java:42)
        at org.apache.iceberg.avro.AvroCustomOrderSchemaVisitor.visit(AvroCustomOrderSchemaVisitor.java:51)
        at org.apache.iceberg.avro.AvroSchemaUtil.buildAvroProjection(AvroSchemaUtil.java:105)
        at org.apache.iceberg.avro.ProjectionDatumReader.setSchema(ProjectionDatumReader.java:68)
        at org.apache.avro.file.DataFileStream.initialize(DataFileStream.java:132)
        at org.apache.avro.file.DataFileReader.<init>(DataFileReader.java:106)
        at org.apache.avro.file.DataFileReader.<init>(DataFileReader.java:98)
        at org.apache.avro.file.DataFileReader.openReader(DataFileReader.java:66)
        at org.apache.iceberg.avro.AvroIterable.newFileReader(AvroIterable.java:100)
        at org.apache.iceberg.avro.AvroIterable.iterator(AvroIterable.java:77)
        at org.apache.iceberg.io.CloseableIterable$3$1.<init>(CloseableIterable.java:95)
        at org.apache.iceberg.io.CloseableIterable$3.iterator(CloseableIterable.java:94)
        at org.apache.iceberg.io.CloseableIterable$3$1.<init>(CloseableIterable.java:95)
        at org.apache.iceberg.io.CloseableIterable$3.iterator(CloseableIterable.java:94)
        at org.apache.iceberg.io.CloseableIterable$3$1.<init>(CloseableIterable.java:95)
        at org.apache.iceberg.io.CloseableIterable$3.iterator(CloseableIterable.java:94)
        at org.apache.iceberg.io.CloseableIterable$3$1.<init>(CloseableIterable.java:95)
        at org.apache.iceberg.io.CloseableIterable$3.iterator(CloseableIterable.java:94)
        at org.apache.iceberg.spark.source.RowDataReader.open(RowDataReader.java:100)
        at org.apache.iceberg.spark.source.BaseDataReader.next(BaseDataReader.java:81)
        at org.apache.spark.sql.execution.datasources.v2.DataSourceRDD$$anon$1.hasNext(DataSourceRDD.scala:49)
        at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.agg_doAggregateWithoutKey_1$(Unknown Source)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.agg_doAggregateWithoutKey_0$(Unknown Source)
        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:636)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:255)
        at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:247)
        at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
        at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$24.apply(RDD.scala:858)
        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:346)
        at org.apache.spark.rdd.RDD.iterator(RDD.scala:310)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
        at org.apache.spark.scheduler.Task.run(Task.scala:123)
        at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

[aokolnychyi]

Can be related to how we handle projections? Reading records works fine.

[shardulm94]

Yes, reading records is fine. It has to do with Spark requesting empty projections for count queries and how we handle those in our Avro readers. A similar issue was fixed in #1077 but it only addressed unpartitioned tables.

[RussellSpitzer]

#1744 I have a ongoing fix for this, @rdblue and I were in dialog about the proper way to fix this and had some disagreements. I do want to get back to it and fix this properly

[KarlManong]

I got the same issue when using version 0.12.1 on spark 3.1.2 executed by kyuubi.

Driver stacktrace:

        at org.apache.spark.scheduler.DAGScheduler.failJobAndIndependentStages(DAGScheduler.scala:2258)

        at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2(DAGScheduler.scala:2207)

        at org.apache.spark.scheduler.DAGScheduler.$anonfun$abortStage$2$adapted(DAGScheduler.scala:2206)

        at scala.collection.mutable.ResizableArray.foreach(ResizableArray.scala:62)

        at scala.collection.mutable.ResizableArray.foreach$(ResizableArray.scala:55)

        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:49)

        at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:2206)

        at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1(DAGScheduler.scala:1079)

        at org.apache.spark.scheduler.DAGScheduler.$anonfun$handleTaskSetFailed$1$adapted(DAGScheduler.scala:1079)

        at scala.Option.foreach(Option.scala:407)

        at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:1079)

        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:2445)

        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2387)

        at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:2376)

        at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:49)

        at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:868)

        at org.apache.spark.SparkContext.runJob(SparkContext.scala:2196)

        at org.apache.spark.SparkContext.runJob(SparkContext.scala:2217)

        at org.apache.spark.SparkContext.runJob(SparkContext.scala:2236)

        at org.apache.spark.SparkContext.runJob(SparkContext.scala:2261)

        at org.apache.spark.rdd.RDD.$anonfun$collect$1(RDD.scala:1030)

        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)

        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)

        at org.apache.spark.rdd.RDD.withScope(RDD.scala:414)

        at org.apache.spark.rdd.RDD.collect(RDD.scala:1029)

        at org.apache.spark.sql.execution.SparkPlan.executeCollect(SparkPlan.scala:390)

        at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3697)

        at org.apache.spark.sql.Dataset.$anonfun$collect$1(Dataset.scala:2966)

        at org.apache.spark.sql.Dataset.$anonfun$withAction$1(Dataset.scala:3688)

        at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$5(SQLExecution.scala:103)

        at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:163)

        at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:90)

        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:775)

        at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:64)

        at org.apache.spark.sql.Dataset.withAction(Dataset.scala:3686)

        at org.apache.spark.sql.Dataset.collect(Dataset.scala:2966)

        at org.apache.kyuubi.engine.spark.operation.ExecuteStatement.$anonfun$executeStatement$1(ExecuteStatement.scala:88)

        at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)

        at org.apache.kyuubi.engine.spark.operation.ExecuteStatement.withLocalProperties(ExecuteStatement.scala:134)

        at org.apache.kyuubi.engine.spark.operation.ExecuteStatement.org$apache$kyuubi$engine$spark$operation$ExecuteStatement$$executeStatement(ExecuteStatement.scala:80)

        at org.apache.kyuubi.engine.spark.operation.ExecuteStatement$$anon$1.run(ExecuteStatement.scala:103)

        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)

        at java.base/java.util.concurrent.FutureTask.run(Unknown Source)

        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)

        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)

        at java.base/java.lang.Thread.run(Unknown Source)

Caused by: java.lang.IllegalArgumentException: Missing required field: data_file

        at org.apache.iceberg.relocated.com.google.common.base.Preconditions.checkArgument(Preconditions.java:217)

        at org.apache.iceberg.avro.BuildAvroProjection.record(BuildAvroProjection.java:99)

        at org.apache.iceberg.avro.BuildAvroProjection.record(BuildAvroProjection.java:43)

        at org.apache.iceberg.avro.AvroCustomOrderSchemaVisitor.visit(AvroCustomOrderSchemaVisitor.java:51)

        at org.apache.iceberg.avro.AvroSchemaUtil.buildAvroProjection(AvroSchemaUtil.java:105)

        at org.apache.iceberg.avro.ProjectionDatumReader.setSchema(ProjectionDatumReader.java:68)

        at org.apache.iceberg.shaded.org.apache.avro.file.DataFileStream.initialize(DataFileStream.java:133)

        at org.apache.iceberg.shaded.org.apache.avro.file.DataFileReader.<init>(DataFileReader.java:130)

        at org.apache.iceberg.shaded.org.apache.avro.file.DataFileReader.<init>(DataFileReader.java:122)

        at org.apache.iceberg.shaded.org.apache.avro.file.DataFileReader.openReader(DataFileReader.java:66)

        at org.apache.iceberg.avro.AvroIterable.newFileReader(AvroIterable.java:100)

        at org.apache.iceberg.avro.AvroIterable.iterator(AvroIterable.java:77)

        at org.apache.iceberg.io.CloseableIterable$4$1.<init>(CloseableIterable.java:99)

        at org.apache.iceberg.io.CloseableIterable$4.iterator(CloseableIterable.java:98)

        at org.apache.iceberg.io.CloseableIterable.lambda$filter$0(CloseableIterable.java:79)

        at org.apache.iceberg.io.CloseableIterable$2.iterator(CloseableIterable.java:73)

        at org.apache.iceberg.io.CloseableIterable$4$1.<init>(CloseableIterable.java:99)

        at org.apache.iceberg.io.CloseableIterable$4.iterator(CloseableIterable.java:98)

        at org.apache.iceberg.ManifestReader.iterator(ManifestReader.java:233)

        at org.apache.iceberg.io.CloseableIterable$4$1.<init>(CloseableIterable.java:99)

        at org.apache.iceberg.io.CloseableIterable$4.iterator(CloseableIterable.java:98)

        at org.apache.iceberg.io.CloseableIterable$4$1.<init>(CloseableIterable.java:99)

        at org.apache.iceberg.io.CloseableIterable$4.iterator(CloseableIterable.java:98)

        at org.apache.iceberg.io.CloseableIterable$4.iterator(CloseableIterable.java:90)

        at org.apache.iceberg.util.Filter.lambda$filter$0(Filter.java:35)

        at org.apache.iceberg.io.CloseableIterable$2.iterator(CloseableIterable.java:73)

        at org.apache.iceberg.spark.source.RowDataReader.open(RowDataReader.java:78)

        at org.apache.iceberg.spark.source.BaseDataReader.next(BaseDataReader.java:93)

        at org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:79)

        at org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:112)

        at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)

        at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)

        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.agg_doAggregateWithoutKey_1$(Unknown Source)

        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.agg_doAggregateWithoutKey_0$(Unknown Source)

        at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)

        at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)

        at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:755)

        at org.apache.spark.sql.execution.SparkPlan.$anonfun$getByteArrayRdd$1(SparkPlan.scala:345)

        at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:898)

        at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:898)

        at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)

        at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:373)

        at org.apache.spark.rdd.RDD.iterator(RDD.scala:337)

        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)

        at org.apache.spark.scheduler.Task.run(Task.scala:131)

        at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:498)

        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1439)

        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:501)

        ... 3 more (state=,code=0)

[KarlManong]

A very brief Test is here:

/*
 * 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.iceberg.spark.sql;

import java.util.Map;
import org.apache.iceberg.spark.SparkCatalogTestBase;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.jupiter.api.Assertions;

import static org.hamcrest.CoreMatchers.instanceOf;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.fail;

public class TestPartitionedTable extends SparkCatalogTestBase {
  public TestPartitionedTable(String catalogName, String implementation, Map<String, String> config) {
    super(catalogName, implementation, config);
  }

  @Before
  public void createTables() {
    sql(
        "CREATE TABLE %s (id bigint, data string, create_date date) USING iceberg PARTITIONED BY (days(create_date))",
        tableName);
    sql(
        "INSERT INTO %s VALUES (1, 'a', date '2021-11-11'), (2, 'b', date '2021-11-11'), (3, 'c', date '2021-11-11')",
        tableName);
  }

  @After
  public void removeTables() {
    sql("DROP TABLE IF EXISTS %s", tableName);
  }

  @Test
  public void testCount() {
    try {
      sql("SELECT count(1) FROM %s.files", tableName);
      fail("No exception!");
    } catch (Exception e) {
      Throwable cause = e.getCause();
      assertThat(cause, instanceOf(IllegalArgumentException.class));
      Assertions.assertEquals("Missing required field: data_file", cause.getMessage());
    }
  }
}

[KarlManong]

@RussellSpitzer Could you please have a look?

[RussellSpitzer]

Should be fixed with #2953

[rdblue]

@KarlManong, can you try with the current master?

[KarlManong]

Should be fixed with #2953

confirmed.

Thank U !