Skip to content

input_file_name() returns empty values above a converted typed Dataset operation #6708

Description

@andygrove

Describe the bug

With spark.comet.convert.typedDataset.enabled=true, input_file_name(), input_file_block_start() and input_file_block_length() return "", -1 and -1 for every row when Spark evaluates them above a Comet operator that reads the converted output of a typed Dataset operation. Spark returns each row's file path and block offsets. The query succeeds, so the wrong values are silent.

This is #6573 for the conversion that #6564 added. CometSparkToColumnarExec fills an Arrow batch from SerializeFromObjectExec, which pulls rows from the scan below the typed operation. By the time the Spark Project evaluates the expressions, the reader has moved on or unset InputFileBlockHolder. The SerializeFromObjectExec case in CometExecRule.transform calls convertTypedDatasetOutput without going through shouldApplySparkToColumnar, so the guard that #6703 adds there does not cover it.

*(2) Project [input_file_name() AS input_file_name()#576, input_file_block_start() AS input_file_block_start()#577L, input_file_block_length() AS input_file_block_length()#578L, value#568L]
+- *(2) CometColumnarToRow
   +- CometFilter [value#568L], (value#568L >= 0)
      +- CometSparkRowToColumnar
         +- *(1) SerializeFromObject [input[0, bigint, false] AS value#568L]
            +- *(1) MapElements ..., obj#566: bigint
               +- *(1) DeserializeToObject assertnotnull(id#558L), obj#564: bigint
                  +- *(1) ColumnarToRow
                     +- FileScan parquet [id#558L] Batched: true, ...

Steps to reproduce

spark.range(9000).repartition(3).write.parquet("/tmp/ifn")

spark.conf.set("spark.comet.convert.typedDataset.enabled", "true")
spark.read
  .parquet("/tmp/ifn")
  .as[Long]
  .map(_ + 1)
  .where("value >= 0")
  .selectExpr("input_file_name()", "input_file_block_start()", "input_file_block_length()", "value")
  .show(3, false)

Comet returns ["", -1, -1, <value>] for all 9000 rows.

Expected behavior

Each row reports the file it was read from, as Spark does. Leaving the typed operation's output unconverted when the plan uses these expressions would also be acceptable.

Additional context

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions