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 73867aace94..acff7a20531 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,16 @@ 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, ...) must be mapped by position even + // when orcUseColumnNames is true, 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/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 97ec1cfba8a..09badfa9db6 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 @@ -167,6 +167,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 (the default, orcUseColumnNames + // stays true). 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..c4dc501a361 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 (the default, orcUseColumnNames + // stays true). 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..6e536f77a16 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 (the default, orcUseColumnNames + // stays true). 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..6e536f77a16 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 (the default, orcUseColumnNames + // stays true). 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..6e536f77a16 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 (the default, orcUseColumnNames + // stays true). 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