From 678f2df06d0a78357b38ca7ceb83f59107a1c99d Mon Sep 17 00:00:00 2001 From: beliefer Date: Mon, 6 Jul 2026 17:20:07 +0800 Subject: [PATCH 1/3] [GLUTEN-12436][VL] Map ORC files by position per-file to match Spark OrcUtils, and remove the orcUseColumnNames config --- .../backendsapi/velox/VeloxBackend.scala | 2 +- .../backendsapi/velox/VeloxIteratorApi.scala | 12 +++- .../apache/gluten/config/VeloxConfig.scala | 9 --- .../gluten/execution/FallbackSuite.scala | 36 +++++------ .../gluten/execution/VeloxScanSuite.scala | 9 ++- cpp/velox/config/VeloxConfig.h | 3 +- cpp/velox/utils/ConfigExtractor.cc | 12 +++- docs/velox-configuration.md | 1 - .../apache/gluten/config/GlutenConfig.scala | 13 ++-- .../execution/GlutenHiveSQLQuerySuite.scala | 62 +++++++++++++++++++ .../execution/GlutenHiveSQLQuerySuite.scala | 62 +++++++++++++++++++ .../execution/GlutenHiveSQLQuerySuite.scala | 62 +++++++++++++++++++ .../execution/GlutenHiveSQLQuerySuite.scala | 62 +++++++++++++++++++ .../execution/GlutenHiveSQLQuerySuite.scala | 62 +++++++++++++++++++ 14 files changed, 364 insertions(+), 43 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala index 14150196817..09deec69e49 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala +++ b/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 } diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala index f81b78c71f5..dfbedbeed4a 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala @@ -54,9 +54,17 @@ 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 can + // decide the column-mapping mode per file: files whose physical schema is + // all Hive placeholder names (_col0, ...) -- or when + // orc.force.positional.evolution is set -- must be mapped by position even + // though ORC is read by name by default, matching vanilla Spark's per-file + // behavior in OrcUtils.requestedColumnIds. Without the table schema the + // native reader has nothing to remap file columns to. + // 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) diff --git a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala index 5d5050be513..3aab8d8c9f1 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala +++ b/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.") diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala index d47d4c85aea..889e6394647 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/FallbackSuite.scala +++ b/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) } + } } } } diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxScanSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxScanSuite.scala index 0ba22bf4df6..7139365e5df 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxScanSuite.scala +++ b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxScanSuite.scala @@ -260,9 +260,14 @@ 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 diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h index 5b5e129c342..74a2cd98a0f 100644 --- a/cpp/velox/config/VeloxConfig.h +++ b/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"; diff --git a/cpp/velox/utils/ConfigExtractor.cc b/cpp/velox/utils/ConfigExtractor.cc index 7ee2deae8a8..4c5af76536e 100644 --- a/cpp/velox/utils/ConfigExtractor.cc +++ b/cpp/velox/utils/ConfigExtractor.cc @@ -267,8 +267,16 @@ std::shared_ptr createHiveConnectorSessionC conf->get(kParquetUseColumnNames, true) ? "true" : "false"; configs[facebook::velox::connector::hive::HiveConfig::kAllowInt32NarrowingSession] = conf->get(kAllowInt32Narrowing, true) ? "true" : "false"; - configs[facebook::velox::connector::hive::HiveConfig::kOrcUseColumnNamesSession] = - conf->get(kOrcUseColumnNames, true) ? "true" : "false"; + // ORC/DWRF files are mapped to the requested schema by name by default, + // matching vanilla Spark. Individual files are still mapped by position by + // the native reader when their physical schema is all Hive placeholder names + // (_col0, ...) or when orc.force.positional.evolution is set below. + configs[facebook::velox::connector::hive::HiveConfig::kOrcUseColumnNamesSession] = "true"; + // Forward Spark's orc.force.positional.evolution so the native reader maps + // ORC/DWRF files by position even in name-based mapping (see + // OrcUtils.requestedColumnIds). + configs[facebook::velox::connector::hive::HiveConfig::kOrcForcePositionalEvolutionSession] = + conf->get(kOrcForcePositionalEvolution, false) ? "true" : "false"; configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterPageSizeSession)] = conf->get(kWriteParquetPageSizeBytes, "1MB"); configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterDictionaryPageSizeLimitSession)] = diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md index 8481e8f60c6..75197345989 100644 --- a/docs/velox-configuration.md +++ b/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. | diff --git a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala b/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala index 93fb3888e99..871692ea24c 100644 --- a/gluten-substrait/src/main/scala/org/apache/gluten/config/GlutenConfig.scala +++ b/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 diff --git a/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala b/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala index e0ca4ae441f..e4c3612727f 100644 --- a/gluten-ut/spark33/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala +++ b/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 diff --git a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala index 93b31a02e90..c69e3fe0c9c 100644 --- a/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala +++ b/gluten-ut/spark34/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala @@ -79,6 +79,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 diff --git a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala index 0938f0dd6cb..bbebfce0010 100644 --- a/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala +++ b/gluten-ut/spark35/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala @@ -149,6 +149,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 diff --git a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala index 0938f0dd6cb..bbebfce0010 100644 --- a/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala +++ b/gluten-ut/spark40/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala @@ -149,6 +149,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 diff --git a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala index 0938f0dd6cb..bbebfce0010 100644 --- a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala +++ b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveSQLQuerySuite.scala @@ -149,6 +149,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 From a24d86ddb48d267950dd31869385f4c211ff8c77 Mon Sep 17 00:00:00 2001 From: beliefer Date: Wed, 8 Jul 2026 17:28:27 +0800 Subject: [PATCH 2/3] Update code --- .../backendsapi/velox/VeloxIteratorApi.scala | 13 ++++++------- cpp/velox/utils/ConfigExtractor.cc | 16 +++++++--------- 2 files changed, 13 insertions(+), 16 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala index dfbedbeed4a..73342a3d4b9 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala @@ -54,13 +54,12 @@ 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 can - // decide the column-mapping mode per file: files whose physical schema is - // all Hive placeholder names (_col0, ...) -- or when - // orc.force.positional.evolution is set -- must be mapped by position even - // though ORC is read by name by default, matching vanilla Spark's per-file - // behavior in OrcUtils.requestedColumnIds. Without the table schema the - // native reader has nothing to remap file columns to. + // 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 ( diff --git a/cpp/velox/utils/ConfigExtractor.cc b/cpp/velox/utils/ConfigExtractor.cc index 4c5af76536e..d1744fe78e7 100644 --- a/cpp/velox/utils/ConfigExtractor.cc +++ b/cpp/velox/utils/ConfigExtractor.cc @@ -268,15 +268,13 @@ std::shared_ptr createHiveConnectorSessionC configs[facebook::velox::connector::hive::HiveConfig::kAllowInt32NarrowingSession] = conf->get(kAllowInt32Narrowing, true) ? "true" : "false"; // ORC/DWRF files are mapped to the requested schema by name by default, - // matching vanilla Spark. Individual files are still mapped by position by - // the native reader when their physical schema is all Hive placeholder names - // (_col0, ...) or when orc.force.positional.evolution is set below. - configs[facebook::velox::connector::hive::HiveConfig::kOrcUseColumnNamesSession] = "true"; - // Forward Spark's orc.force.positional.evolution so the native reader maps - // ORC/DWRF files by position even in name-based mapping (see - // OrcUtils.requestedColumnIds). - configs[facebook::velox::connector::hive::HiveConfig::kOrcForcePositionalEvolutionSession] = - conf->get(kOrcForcePositionalEvolution, false) ? "true" : "false"; + // 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(kOrcForcePositionalEvolution, false) ? "false" : "true"; configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterPageSizeSession)] = conf->get(kWriteParquetPageSizeBytes, "1MB"); configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kWriterDictionaryPageSizeLimitSession)] = From c03e48d399f9d858ef39cd38b1c2f42776e83a47 Mon Sep 17 00:00:00 2001 From: beliefer Date: Wed, 8 Jul 2026 17:40:08 +0800 Subject: [PATCH 3/3] Update code --- .../backendsapi/velox/VeloxIteratorApi.scala | 14 ++++++-------- .../apache/gluten/execution/VeloxScanSuite.scala | 7 +++---- 2 files changed, 9 insertions(+), 12 deletions(-) diff --git a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala index 73342a3d4b9..5f623abffc7 100644 --- a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala +++ b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxIteratorApi.scala @@ -54,14 +54,12 @@ 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). + // 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) || (fileFormat == ReadFileFormat.ParquetReadFormat && !VeloxConfig.get.parquetUseColumnNames) diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxScanSuite.scala b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxScanSuite.scala index 7139365e5df..f51b7d72403 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxScanSuite.scala +++ b/backends-velox/src/test/scala/org/apache/gluten/execution/VeloxScanSuite.scala @@ -262,10 +262,9 @@ class VeloxScanSuite extends VeloxWholeStageTransformerSuite { 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. + // `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( GlutenConfig.SPARK_ORC_FORCE_POSITIONAL_EVOLUTION -> "true") { withTempDir {