[Experimental scans] schema adapter does not apply required schema for structs within lists

[andygrove]

Describe the bug

The following test currently fails:

  test("nested data - array of struct") {
    val data = (1 to 10).map(i => Tuple1(Seq(i -> s"val_$i")))
    withParquetTable(data, "t") {
      withSQLConf(CometConf.COMET_NATIVE_SCAN_IMPL.key -> CometConf.SCAN_NATIVE_ICEBERG_COMPAT) {
        checkSparkAnswerAndOperator(sql("SELECT _1[0]._2 FROM t"))
      }
    }
  }

The parquet file contains an array of struct, where the struct has fields _1 and _2.

The required schema in the scan filters out the _1 from the struct:

required_schema = Field { name: "_1", data_type: List(Field { name: "item", data_type: Struct([Field { name: "_2", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }

The projection contains GetStructField with ordinal 0 to represent field _2 in the struct.

Comet does not apply the required schema and therefore selects both _1 and _2 from the struct, and the GetStructField with ordinal=0 returns _1 instead of _2.

Steps to reproduce

No response

Expected behavior

No response

Additional context

No response

[comphead]

another test to reproduce

  test("native reader - read a STRUCT subfield from ARRAY of STRUCTS") {
    testSingleLineQuery(
      """
        | select array(str0, str1) c0 from
        | (
        |   select named_struct('a', 1, 'b', 'n') str0, named_struct('a', 2, 'b', 'w') str1
        | )
        |""".stripMargin,
      "select c0[0].b col0 from tbl")
  }

interesting that referring to first column which is a returns result correctly, but b does not

[andygrove]

interesting that referring to first column which is a returns result correctly, but b does not

This is because we read all fields in the struct, so we are guaranteed to produce the correct results in the following cases:

• the user only selects the first field
• the user selects all fields

[andygrove]

interesting that referring to first column which is a returns result correctly, but b does not

This is because we read all fields in the struct, so we are guaranteed to produce the correct results in the following cases:

* the user only selects the first field

* the user selects all fields

That is not quite correct. We will accidentally have the correct behavior when the user selects a set of fields starting with the first field and not skipping any fields in that range. So for a struct with fields a, b, c, d, we will have correct behavior for reading a or a, b or a, b, c, or a, b, c, d.

[comphead]

interesting that referring to first column which is a returns result correctly, but b does not

This is because we read all fields in the struct, so we are guaranteed to produce the correct results in the following cases:

* the user only selects the first field

* the user selects all fields

That is not quite correct. We will accidentally have the correct behavior when the user selects a set of fields starting with the first field and not skipping any fields in that range. So for a struct with fields a, b, c, d, we will have correct behavior for reading a or a, b or a, b, c, or a, b, c, d.

Reverse also works. but a, c skipping b doesn't

PASSED
test("native reader - read a STRUCT subfield from ARRAY of STRUCTS - field from first") {
    testSingleLineQuery(
      """
        | select array(str0, str1) c0 from
        | (
        |   select
        |     named_struct('a', 1, 'b', 'n', 'c', 'x') str0,
        |     named_struct('a', 2, 'b', 'w', 'c', 'y') str1
        | )
        |""".stripMargin,
      "select c0[0].a, c0[0].b, c0[0].c from tbl")
  }


  PASSED
  test("native reader - read a STRUCT subfield from ARRAY of STRUCTS - reverse fields") {
    testSingleLineQuery(
      """
        | select array(str0, str1) c0 from
        | (
        |   select
        |     named_struct('a', 1, 'b', 'n', 'c', 'x') str0,
        |     named_struct('a', 2, 'b', 'w', 'c', 'y') str1
        | )
        |""".stripMargin,
      "select c0[0].c, c0[0].b, c0[0].a from tbl")
  }

  FAILED
  test("native reader - read a STRUCT subfield from ARRAY of STRUCTS - skip field") {
    testSingleLineQuery(
      """
        | select array(str0, str1) c0 from
        | (
        |   select
        |     named_struct('a', 1, 'b', 'n', 'c', 'x') str0,
        |     named_struct('a', 2, 'b', 'w', 'c', 'y') str1
        | )
        |""".stripMargin,
      "select c0[0].a, c0[0].c from tbl")
  }

[comphead]

The simplest test

  test("native reader - read a STRUCT subfield - field from second") {
    testSingleLineQuery(
      """
        |select array(named_struct('a', 1, 'b', 'n')) c0
        |""".stripMargin,
      "select c0[0].b from tbl")
  }

[comphead]

Narrowing it down, the correctness issue comes from ListExtract

the GetStructField looks works correctly, I was playing with 2 tests

test("native reader - arr index") {
    withSQLConf(
      CometConf.COMET_EXEC_ENABLED.key -> "true",
      SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
      CometConf.COMET_ENABLED.key -> "true",
      CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "false",
      CometConf.COMET_NATIVE_SCAN_IMPL.key -> "native_datafusion") {
      testSingleLineQuery(
        """
          | select array(str0) c0 from
          | (
          |   select
          |     named_struct('a', 1, 'b', 'n', 'c', 'x') str0
          | )
          |""".stripMargin,
        "select c0[0].a, c0[0].c from tbl",
        debugCometDF = df => df.explain("formatted"))
    }
  }

  test("native reader - struct") {
    withSQLConf(
      CometConf.COMET_EXEC_ENABLED.key -> "true",
      SQLConf.USE_V1_SOURCE_LIST.key -> "parquet",
      CometConf.COMET_ENABLED.key -> "true",
      CometConf.COMET_EXPLAIN_FALLBACK_ENABLED.key -> "false",
      CometConf.COMET_NATIVE_SCAN_IMPL.key -> "native_datafusion") {
      testSingleLineQuery(
        """
          | select str0 c0 from
          | (
          |   select
          |     named_struct('a', 1, 'b', 'n', 'c', 'x') str0
          | )
          |""".stripMargin,
        "select c0.a, c0.c from tbl",
        debugCometDF = df => df.explain("formatted"))
    }
  }

And I can see that during evaluate the record batch already comes with wrong data, might be related to projection pushdown, investigating further

[comphead]

I went down the call stack and found the reader gives batches wrong in OpStruct::NativeScan(scan)

[core/src/execution/planner.rs:1169:17] result = [
    Ok(
        RecordBatch {
            schema: Schema {
                fields: [
                    Field {
                        name: "c0",
                        data_type: List(
                            Field {
                                name: "item",
                                data_type: Struct(
                                    [
                                        Field {
                                            name: "a",
                                            data_type: Int32,
                                            nullable: true,
                                            dict_id: 0,
                                            dict_is_ordered: false,
                                            metadata: {},
                                        },
                                        Field {
                                            name: "c",
                                            data_type: Utf8,
                                            nullable: true,
                                            dict_id: 0,
                                            dict_is_ordered: false,
                                            metadata: {},
                                        },
                                    ],
                                ),
                                nullable: true,
                                dict_id: 0,
                                dict_is_ordered: false,
                                metadata: {},
                            },
                        ),
                        nullable: true,
                        dict_id: 0,
                        dict_is_ordered: false,
                        metadata: {},
                    },
                ],
                metadata: {},
            },
            columns: [
                ListArray
                [
                  StructArray
                -- validity:
                [
                  valid,
                ]
                [
                -- child 0: "a" (Int32)
                PrimitiveArray<Int32>
                [
                  1,
                ]
                -- child 1: "c" (Utf8)
                StringArray
                [
                  "n",
                ]
                ],
                ],
            ],
            row_count: 1,
        },
    ),
]

Value for c column is n from reader however it should be x as the n it is a column B, checking

[comphead]

Possible problem is in SparkSchemaAdapterFactory