[GLUTEN-12436][VL] Map ORC files by position per-file to match Spark's OrcUtils, and remove the orcUseColumnNames config

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala

@@ -233,7 +233,7 @@ object VeloxBackendSettings extends BackendSettingsApi {
     }
 
     def validateDataSchema(): Option[String] = {
-      if (VeloxConfig.get.parquetUseColumnNames && VeloxConfig.get.orcUseColumnNames) {
+      if (VeloxConfig.get.parquetUseColumnNames) {
         return None
       }
 

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala

@@ -54,9 +54,14 @@ class VeloxIteratorApi extends IteratorApi with Logging {
       localFilesNode: LocalFilesNode,
       fileSchema: StructType,
       fileFormat: ReadFileFormat): LocalFilesNode = {
+    // For ORC/DWRF, always attach the table schema so the native reader has a target to remap file
+    // columns to. It is needed both when the whole scan is mapped by position
+    // (orc.force.positional.evolution) and for the per-file fallback where a file whose physical
+    // schema is all Hive placeholder names (_col0, ...) is mapped by position even though ORC is
+    // read by name by default, matching vanilla Spark's OrcUtils.requestedColumnIds.
+    // For Parquet, keep the previous behavior (only attach when mapping by position).
     if (
-      ((fileFormat == ReadFileFormat.OrcReadFormat || fileFormat == ReadFileFormat.DwrfReadFormat)
-        && !VeloxConfig.get.orcUseColumnNames)
+      (fileFormat == ReadFileFormat.OrcReadFormat || fileFormat == ReadFileFormat.DwrfReadFormat)
       || (fileFormat == ReadFileFormat.ParquetReadFormat && !VeloxConfig.get.parquetUseColumnNames)
     ) {
       localFilesNode.setFileSchema(fileSchema)

backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala

@@ -97,9 +97,6 @@ class VeloxConfig(conf: SQLConf) extends GlutenConfig(conf) {
 
   def cudfShuffleMaxPrefetchBytes: Long = getConf(CUDF_SHUFFLE_MAX_PREFETCH_BYTES)
 
-  def orcUseColumnNames: Boolean = getConf(ORC_USE_COLUMN_NAMES) &&
-    !conf.getConfString(GlutenConfig.SPARK_ORC_FORCE_POSITIONAL_EVOLUTION, "false").toBoolean
-
   def parquetUseColumnNames: Boolean = getConf(PARQUET_USE_COLUMN_NAMES)
 
   def parquetPageSizeBytes: Long = getConf(PARQUET_PAGE_SIZE_BYTES)
@@ -870,12 +867,6 @@ object VeloxConfig extends ConfigRegistry {
       .intConf
       .createWithDefault(100)
 
-  val ORC_USE_COLUMN_NAMES =
-    buildConf("spark.gluten.sql.columnar.backend.velox.orcUseColumnNames")
-      .doc("Maps table field names to file field names using names, not indices for ORC files.")
-      .booleanConf
-      .createWithDefault(true)
-
   val PARQUET_USE_COLUMN_NAMES =
     buildConf("spark.gluten.sql.columnar.backend.velox.parquetUseColumnNames")
       .doc("Maps table field names to file field names using names, not indices for Parquet files.")

backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala

@@ -316,27 +316,23 @@ class FallbackSuite extends VeloxWholeStageTransformerSuite with AdaptiveSparkPl
       format =>
         Seq("true", "false").foreach {
           parquetUseColumnNames =>
-            Seq("true", "false").foreach {
-              orcUseColumnNames =>
-                withSQLConf(
-                  VeloxConfig.PARQUET_USE_COLUMN_NAMES.key -> parquetUseColumnNames,
-                  VeloxConfig.ORC_USE_COLUMN_NAMES.key -> orcUseColumnNames
-                ) {
-                  withTable("test") {
-                    spark
-                      .range(100)
-                      .selectExpr("to_timestamp_ntz(from_unixtime(id % 3)) as c1", "id as c2")
-                      .write
-                      .format(format)
-                      .saveAsTable("test")
-
-                    runQueryAndCompare(query) {
-                      df =>
-                        val plan = df.queryExecution.executedPlan
-                        assert(collect(plan) { case g: GlutenPlan => g }.nonEmpty)
-                    }
-                  }
+            withSQLConf(
+              VeloxConfig.PARQUET_USE_COLUMN_NAMES.key -> parquetUseColumnNames
+            ) {
+              withTable("test") {
+                spark
+                  .range(100)
+                  .selectExpr("to_timestamp_ntz(from_unixtime(id % 3)) as c1", "id as c2")
+                  .write
+                  .format(format)
+                  .saveAsTable("test")
+
+                runQueryAndCompare(query) {
+                  df =>
+                    val plan = df.queryExecution.executedPlan
+                    assert(collect(plan) { case g: GlutenPlan => g }.nonEmpty)
                 }
+              }
             }
         }
     }

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxScanSuite.scala

@@ -260,9 +260,13 @@ class VeloxScanSuite extends VeloxWholeStageTransformerSuite {
     }
   }
 
-  test("ORC index based schema evolution") {
+  test("ORC positional schema evolution") {
+    // Vanilla Spark maps ORC columns by position (not by name) only when
+    // `orc.force.positional.evolution=true` or the file's physical names are all Hive placeholders
+    // (see OrcUtils.requestedColumnIds). Here the file has real names (a, b) but the table uses
+    // different names (c, d, e); positional mapping is forced so c<-a, d<-b, e<-missing.
     withSQLConf(
-      VeloxConfig.ORC_USE_COLUMN_NAMES.key -> "false") {
+      GlutenConfig.SPARK_ORC_FORCE_POSITIONAL_EVOLUTION -> "true") {
       withTempDir {
         dir =>
           val path = dir.getCanonicalPath

cpp/velox/config/VeloxConfig.h

@@ -175,7 +175,8 @@ const std::string kMaxCoalescedBytes = "spark.gluten.sql.columnar.backend.velox.
 const std::string kCachePrefetchMinPct = "spark.gluten.sql.columnar.backend.velox.cachePrefetchMinPct";
 const std::string kMemoryPoolCapacityTransferAcrossTasks =
     "spark.gluten.sql.columnar.backend.velox.memoryPoolCapacityTransferAcrossTasks";
-const std::string kOrcUseColumnNames = "spark.gluten.sql.columnar.backend.velox.orcUseColumnNames";
+const std::string kOrcForcePositionalEvolution =
+    "spark.gluten.sql.columnar.backend.velox.orcForcePositionalEvolution";
 const std::string kParquetUseColumnNames = "spark.gluten.sql.columnar.backend.velox.parquetUseColumnNames";
 const std::string kAllowInt32Narrowing = "spark.gluten.sql.columnar.backend.velox.allowInt32Narrowing";
 

cpp/velox/utils/ConfigExtractor.cc

@@ -267,8 +267,14 @@ std::shared_ptr<facebook::velox::config::ConfigBase> createHiveConnectorSessionC
       conf->get<bool>(kParquetUseColumnNames, true) ? "true" : "false";
   configs[facebook::velox::connector::hive::HiveConfig::kAllowInt32NarrowingSession] =
       conf->get<bool>(kAllowInt32Narrowing, true) ? "true" : "false";
+  // ORC/DWRF files are mapped to the requested schema by name by default,
+  // matching vanilla Spark. Individual files whose physical schema is all Hive
+  // placeholder names (_col0, ...) are still mapped by position per-file by the
+  // native reader. When Spark's orc.force.positional.evolution is set, force
+  // position-based mapping for the whole scan by disabling name-based mapping
+  // (ColumnMappingMode::kPosition), matching OrcUtils.requestedColumnIds.
   configs[facebook::velox::connector::hive::HiveConfig::kOrcUseColumnNamesSession] =
-      conf->get<bool>(kOrcUseColumnNames, true) ? "true" : "false";
+      conf->get<bool>(kOrcForcePositionalEvolution, false) ? "false" : "true";
   configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterPageSizeSession)] =
       conf->get<std::string>(kWriteParquetPageSizeBytes, "1MB");
   configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterDictionaryPageSizeLimitSession)] =

docs/velox-configuration.md

@@ -56,7 +56,6 @@ nav_order: 16
 | spark.gluten.sql.columnar.backend.velox.memoryPoolCapacityTransferAcrossTasks    | 🔄 Dynamic    | true              | Whether to allow memory capacity transfer between memory pools from different tasks.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
 | spark.gluten.sql.columnar.backend.velox.memoryUseHugePages                       | 🔄 Dynamic    | false             | Use explicit huge pages for Velox memory allocation.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
 | spark.gluten.sql.columnar.backend.velox.orc.scan.enabled                         | 🔄 Dynamic    | true              | Enable velox orc scan. If disabled, vanilla spark orc scan will be used.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  |
-| spark.gluten.sql.columnar.backend.velox.orcUseColumnNames                        | 🔄 Dynamic    | true              | Maps table field names to file field names using names, not indices for ORC files.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
 | spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes          | 🔄 Dynamic    | 2MB               | The maximum size in bytes for a Parquet dictionary page                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
 | spark.gluten.sql.columnar.backend.velox.parquet.pageSizeBytes                    | 🔄 Dynamic    | 1MB               | The page size in bytes is for compression.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
 | spark.gluten.sql.columnar.backend.velox.parquetMaxTargetFileSize                 | 🔄 Dynamic    | 0b                | The target file size for each output file when writing data. 0 means no limit on target file size, and the actual file size will be determined by other factors such as max partition number and shuffle batch size.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |

gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala

@@ -597,16 +597,19 @@ object GlutenConfig extends ConfigRegistry {
       .foreach { case (k, v) => nativeConfMap.put(k, v) }
 
     // When `orc.force.positional.evolution=true`, vanilla Spark maps ORC columns by
-    // position rather than by name (see OrcUtils.requestedColumnIds). The Velox ORC reader
-    // must do the same, otherwise name-based matching against a mismatched file schema
-    // reads columns back as null/empty. Override the (Velox) orcUseColumnNames session conf
-    // so native reads ORC by position too. Harmless for backends that ignore this key.
+    // position rather than by name (see OrcUtils.requestedColumnIds). Forward the flag to
+    // the native (Velox) reader so it maps ORC/DWRF files by position too, otherwise
+    // name-based matching against a mismatched file schema reads columns back as null/empty.
+    // The native reader still decides per file (files with all-`_col*` physical names are
+    // always mapped by position). Harmless for backends that ignore this key.
     // String literal is used because gluten-substrait cannot depend on backends-velox.
     if (
       backendName == "velox" &&
       conf.getOrElse(SPARK_ORC_FORCE_POSITIONAL_EVOLUTION, "false").toBoolean
     ) {
-      nativeConfMap.put("spark.gluten.sql.columnar.backend.velox.orcUseColumnNames", "false")
+      nativeConfMap.put(
+        "spark.gluten.sql.columnar.backend.velox.orcForcePositionalEvolution",
+        "true")
     }
 
     // Pass the latest tokens to native

gluten-ut/spark33/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala

@@ -166,6 +166,68 @@ class GlutenHiveSQLQuerySuite extends GlutenHiveSQLQuerySuiteBase {
     }
   }
 
+  testGluten(
+    "GLUTEN: Hive ORC files with _col* names read by position without positional flag") {
+    // Regression for the case where two ORC tables must use OPPOSITE column
+    // mapping modes in the same query: one with real column names (by name) and
+    // one written by old Hive with placeholder _col* names (by position). The
+    // native reader must decide the mode per file (matching vanilla Spark's
+    // OrcUtils.requestedColumnIds), so a _col* file reads correctly even though
+    // orc.force.positional.evolution is NOT set (ORC is read by name by
+    // default). Without the fix the _col* columns would read back as NULL.
+    val hiveClient: HiveClient =
+      spark.sharedState.externalCatalog.unwrapped.asInstanceOf[HiveExternalCatalog].client
+
+    withSQLConf("spark.sql.hive.convertMetastoreOrc" -> "false") {
+      withTempDir {
+        dir =>
+          val colStarLoc = s"file:///$dir/test_orc_colstar"
+          val namedLoc = s"file:///$dir/test_orc_named"
+          withTable("test_orc_colstar", "test_orc_colstar_renamed", "test_orc_named") {
+            // Naming the columns literally _col0/_col1 guarantees the physical
+            // ORC field names are placeholders, independent of the Hive
+            // version (mirrors Spark's SPARK-34897 setup).
+            hiveClient.runSqlHive(
+              s"create table test_orc_colstar(_col0 int, _col1 string) " +
+                s"stored as orc location '$colStarLoc'")
+            hiveClient.runSqlHive("insert into test_orc_colstar select 7, 'a'")
+
+            // A second table over the SAME files but with real names. By name,
+            // id/name are absent from the _col* files; only position mapping
+            // can read them -- and it must happen WITHOUT the positional flag.
+            hiveClient.runSqlHive(
+              s"create table test_orc_colstar_renamed(id int, name string) " +
+                s"stored as orc location '$colStarLoc'")
+
+            // A table with real physical column names, read by name.
+            hiveClient.runSqlHive(
+              s"create table test_orc_named(uid int, label string) " +
+                s"stored as orc location '$namedLoc'")
+            hiveClient.runSqlHive("insert into test_orc_named select 7, 'b'")
+
+            // No positional flag set. The _col* table read via real names must
+            // still return the values (positional fallback).
+            val colStar = sql("select id, name from test_orc_colstar_renamed")
+            checkAnswer(colStar, Seq(Row(7, "a")))
+            checkOperatorMatch[HiveTableScanExecTransformer](colStar)
+
+            // The real-name table still reads correctly by name in the same
+            // session (opposite mapping mode).
+            val named = sql("select uid, label from test_orc_named")
+            checkAnswer(named, Seq(Row(7, "b")))
+            checkOperatorMatch[HiveTableScanExecTransformer](named)
+
+            // Both in one query (the original failure folded the join to an
+            // empty LocalTableScan). The join must return a non-empty result.
+            val joined = sql(
+              "select c.name, n.label from test_orc_colstar_renamed c " +
+                "join test_orc_named n on c.id = n.uid")
+            checkAnswer(joined, Seq(Row("a", "b")))
+          }
+      }
+    }
+  }
+
   test("GLUTEN-11062: Supports mixed input format for partitioned Hive table") {
     val hiveClient: HiveClient =
       spark.sharedState.externalCatalog.unwrapped.asInstanceOf[HiveExternalCatalog].client