Fix file format checks to be exact and handle Delta Lake column mapping [databricks]

[jlowe]

Fixes #9255.

This fixes the following problems:

• Format checks were allowing derived classes of the CPU class to be considered supported
• DeltaParquetFileFormat semantics were being completely ignored (i.e.; column mapping and deletion vectors)

File format checks were updated to check for specific CPU classes so we don't accidentally think we know how to replace the semantics when a derived class that alters the behavior is used instead.

The interface for Delta Lake providers was updated to handle read interfaces, and ExternalSource was updated accordingly.

[jlowe]

build

[revans2]

Looks good just one question, and there are some test failures so I am not going to approve it yet.

[revans2]

When would we ever have replaced the file format and be tagging the plan again? I understand if we are just being cautious, but with the other checks for an exact class it feels like we would have made some horrible franken mix of CPU exec and GPU FileFormat to hit this. That or I don't really understand the context that this is called in. If that is the case, then we need some better docs for the method definition.

[jlowe]

This checks for the GPU version of the format because isSupportedFormat is how ExternalSource figures out which source, of many, is supposed to handle a particular call. For example, one of the interfaces is createMultiFileReaderFactory and that happens after we've already converted to a GPU format. I'll add some comments to the various isSupported methods of the provider interfaces to make this clearer.

[revans2]

Okay could you add some docs to the isSupportedFormat declaration about who could call this and what is expected.

[jlowe]

I decided this API is inherently confusing and fragile given it's looking at non-CPU formats. I ended up solving this via #9283 and will update this PR to be based on that. I'll also update the isSupportedFormat calls to check for the explicit CPU class we're expecting.

[jlowe]

Putting this in draft until it's been upmerged with #9283.

[LIN-Yu-Ting]

@jlowe. Very interesting development. I saw that

1. you modified GpuFileSourceScanExec.scala to handle DeltaParquetFileFormat with ExternalSource.scala.
2. DeltaProvider inside ExternalSource, implemented by DeltaProviderImpl.scala will execute function createMultiFileReaderFactory.
3. Inside, you will run createMultiFileReaderFactory based on delta table version and finally all of them call GpuDeltaParquetFileFormat to return ReaderFactory.

At the end, you prepare metadata such as StructType for DeltaTable with columnMappingMode.

    GpuParquetMultiFilePartitionReaderFactory(
      fileScan.conf,
      broadcastedConf,
      prepareSchema(fileScan.relation.dataSchema),
      prepareSchema(fileScan.requiredSchema),
      prepareSchema(fileScan.readPartitionSchema),
      pushedFilters,
      fileScan.rapidsConf,
      fileScan.allMetrics,
      fileScan.queryUsesInputFile,
      fileScan.alluxioPathsMap)

which is different from original return value inside GpuFileSourceScanExec.

      GpuParquetMultiFilePartitionReaderFactory(
        sqlConf,
        broadcastedHadoopConf,
        relation.dataSchema,
        requiredSchema,
        readPartitionSchema,
        pushedDownFilters.toArray,
        rapidsConf,
        allMetrics,
        queryUsesInputFile,
        alluxioPathReplacementMap)

[jlowe]

you prepare metadata such as StructType for DeltaTable with columnMappingMode

Yes, this reflects the behavior implemented by DeltaParquetFileFormat, the format used by the CPU when reading Delta Lake tables. See https://github.com/delta-io/delta/blob/master/spark/src/main/scala/org/apache/spark/sql/delta/DeltaParquetFileFormat.scala#L128-L136

[jlowe]

build