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..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,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) 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 0dc6f052e0c..b742e3a5a01 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) @@ -903,12 +900,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 47944379107..2a1fd3bf2b2 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 @@ -284,9 +284,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 diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h index 36d029e8ff4..b6dd5f9fa00 100644 --- a/cpp/velox/config/VeloxConfig.h +++ b/cpp/velox/config/VeloxConfig.h @@ -184,7 +184,7 @@ 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 240033badb0..109cf7b8077 100644 --- a/cpp/velox/utils/ConfigExtractor.cc +++ b/cpp/velox/utils/ConfigExtractor.cc @@ -267,8 +267,14 @@ std::shared_ptr createHiveConnectorSessionC conf->get(kParquetUseColumnNames, true) ? "true" : "false"; 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 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(kOrcUseColumnNames, true) ? "true" : "false"; + conf->get(kOrcForcePositionalEvolution, false) ? "false" : "true"; 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 4787b075543..9ef1b3ba4b3 100644 --- a/docs/velox-configuration.md +++ b/docs/velox-configuration.md @@ -61,7 +61,6 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.memoryUseHugePages | 🔄 Dynamic | false | Use explicit huge pages for Velox memory allocation. | | spark.gluten.sql.columnar.backend.velox.numCacheFileHandles | ⚓ Static | 10000 | Maximum number of entries in the file handle cache. Each entry holds an open file descriptor (local FS) or connection state (remote FS). Note that on local filesystems, high values may approach the OS file descriptor limit (ulimit -n). On remote object stores (S3, ABFS, GCS) entries represent network connections/sockets rather than per-file OS file descriptors, but they can still count toward OS resource limits (ulimit -n). | | 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 e4fafa35999..8ffb6345c39 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 @@ -593,16 +593,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..d7d6c57c5b0 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..010e55c346a 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..5308ee7ede3 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..5308ee7ede3 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..5308ee7ede3 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