Fallback Delta file scans to CPU on DB-17.3 [databricks]

[nartal1]

Fixes #14607

Description

• On DB-17.3, UC managed tables use DeltaParquetFileFormat which requires
  Delta-specific columns (e.g. skip_row). Since there is no Delta provider
  for DB-17.3, the GPU reader cannot fill these columns.
• Forked FileSourceScanExecMeta for the 400db173 shim to detect unsupported
  ParquetFileFormat subclasses and fall back the scan to CPU.
• Created FileSourceScanExecMetaBase to share the common logic across DB shims.
• Plain Parquet reads are unaffected.

Tested this manually on DB cluster

• UC managed Delta table on DB-17.3 — scan on CPU, rest on GPU
• Raw Parquet read on DB-17.3 — scan stays on GPU

Checklists

Documentation

• Updated for new or modified user-facing features or behaviors
• No user-facing change

Testing

• Added or modified tests to cover new code paths
• Covered by existing tests
  (Please provide the names of the existing tests in the PR description.)
• Not required

Performance

• Tests ran and results are added in the PR description
• Issue filed with a link in the PR description
• Not required

[greptile-apps]

Greptile Summary

This PR falls back FileSourceScanExec to CPU on DB-17.3 when the file format is an unsupported ParquetFileFormat subclass (e.g. DeltaParquetFileFormat) by forking FileSourceScanExecMeta for the 400db173 shim and extracting shared logic into a new FileSourceScanExecMetaBase abstract class. The approach is correct: GpuFileSourceScanExec.tagSupport treats every ParquetFileFormat subclass as GPU-compatible, so the explicit override adds the necessary willNotWorkOnGpu gate before convertToGpu can be reached.

Confidence Score: 5/5

Safe to merge — logic is correct, shim annotations are consistent, and the fallback gate is properly applied.

All three files are clean: the base class extraction is mechanical, the 330db shim correctly drops to a single-line delegate, and the 400db173 override correctly intercepts DeltaParquetFileFormat subclasses before convertToGpu is reached. No P0/P1 findings.

No files require special attention.

Important Files Changed

Filename	Overview
sql-plugin/src/main/spark330db/scala/com/nvidia/spark/rapids/shims/FileSourceScanExecMetaBase.scala	New abstract base class extracted from the original FileSourceScanExecMeta; carries the full shim annotation including 400db173, placed in the lowest buildver directory per project convention.
sql-plugin/src/main/spark330db/scala/com/nvidia/spark/rapids/shims/FileSourceScanExecMeta.scala	Simplified to a single-line subclass of FileSourceScanExecMetaBase; 400db173 correctly removed from its shim annotation.
sql-plugin/src/main/spark400db173/scala/com/nvidia/spark/rapids/shims/FileSourceScanExecMeta.scala	New DB-17.3-specific override that adds a willNotWorkOnGpu gate for unrecognized ParquetFileFormat subclasses; logic is sound and correctly excludes plain ParquetFileFormat and ExternalSource-supported formats.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A[FileSourceScanExec encountered] --> B{Shim version?}
    B -- "330db / 332db / 341db / 350db143" --> C[FileSourceScanExecMeta\n spark330db]
    B -- "400db173" --> D[FileSourceScanExecMeta\n spark400db173]

    C --> E[FileSourceScanExecMetaBase\n tagPlanForGpu]
    D --> F[super.tagPlanForGpu\n FileSourceScanExecMetaBase]
    F --> E

    E --> G{Delta JSON index?}
    G -- yes --> H[entirePlanWillNotWork → CPU]
    G -- no --> I[ScanExecShims.tagGpuFileSourceScanExecSupport]

    I --> J{format class}
    J -- ParquetFileFormat subclass --> K[GpuReadParquetFileFormat.tagSupport\n passes for DeltaParquetFileFormat!]
    J -- CSV/ORC/JSON --> L[respective GPU tagSupport]
    J -- ExternalSource supported --> M[ExternalSource.tagSupportForGpuFileSourceScan]

    K --> N{400db173 extra check}
    D --> N
    N -- "isAssignableFrom ParquetFileFormat\n && != ParquetFileFormat\n && !ExternalSource.isSupportedFormat" --> O[willNotWorkOnGpu → CPU fallback]
    N -- plain ParquetFileFormat --> P[GPU scan proceeds]

Loading

_{Reviews (2): Last reviewed commit: "Update sql-plugin/src/main/spark330db/sc..." | Re-trigger Greptile}

[nartal1]

Verified with the simple query on DB:

spark.sql("CREATE TABLE rapids_uc_repro_test (id INT, name STRING)")
spark.sql("INSERT INTO rapids_uc_repro_test VALUES (1, 'A'), (2, 'B'), (3, 'C')")

spark.table("unity_catalog_test_workspace.default.rapids_uc_repro_test").explain()
spark.table("unity_catalog_test_workspace.default.rapids_uc_repro_test").show()

Output:

== Physical Plan ==
GpuColumnarToRow false
+- GpuProject [id#101, name#102], true
   +- GpuCoalesceBatches targetsize(1073741824)
      +- GpuFilter if (gpuisnotnull(_databricks_internal_edge_computed_column_skip_row#208)) (_databricks_internal_edge_computed_column_skip_row#208 = false) else gpuisnotnull(raise_error(DELTA_SKIP_ROW_COLUMN_NOT_FILLED, keys: [], values: []))
         +- HostColumnarToGpu targetsize(1073741824)
            +- FileScan parquet unity_catalog_test_workspace.default.rapids_uc_repro_test[id#101,name#102,_databricks_internal_edge_computed_column_skip_row#208] Batched: true, DataFilters: [], Format: Parquet, Location: PreparedDeltaFileIndex(1 paths)[s3://<DB_WORKSPACE>/unity-catalog/145252..., PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id:int,name:string,_databricks_internal_edge_computed_column_skip_row:boolean>


== Optimizer Statistics (table names per statistics state) ==
  missing = rapids_uc_repro_test
  partial = 
  full    = 
Corrective actions: consider running the following command on all tables with missing or partial statistics
  ANALYZE TABLE <table-name> COMPUTE STATISTICS FOR ALL COLUMNS

+---+-------+
| id|  name|
+---+-------+
|  1|  A|
|  2|  B|
|  3|  C|
+---+-------+

[nartal1]

build

[nartal1]

build

[gerashegalov]

LGTM

[gerashegalov]

build

[pxLi]

build

[pxLi]

build

[pxLi]

build