This is an automated email from the ASF dual-hosted git repository.
rui-mo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new c0637ccada [GLUTEN-12436][VL] Make ORC reader's position/name
selection behavior fully in line with Spark (#12453)
c0637ccada is described below
commit c0637ccadae631410170c38f5512e427267f6921
Author: Jiaan Geng <[email protected]>
AuthorDate: Sat Aug 1 00:13:45 2026 +0800
[GLUTEN-12436][VL] Make ORC reader's position/name selection behavior fully
in line with Spark (#12453)
---
.../gluten/backendsapi/velox/VeloxBackend.scala | 2 +-
.../backendsapi/velox/VeloxIteratorApi.scala | 9 +++-
.../org/apache/gluten/config/VeloxConfig.scala | 9 ----
.../apache/gluten/execution/FallbackSuite.scala | 36 ++++++-------
.../apache/gluten/execution/VeloxScanSuite.scala | 8 ++-
cpp/velox/config/VeloxConfig.h | 2 +-
cpp/velox/utils/ConfigExtractor.cc | 8 ++-
docs/velox-configuration.md | 1 -
.../org/apache/gluten/config/GlutenConfig.scala | 13 +++--
.../hive/execution/GlutenHiveSQLQuerySuite.scala | 62 ++++++++++++++++++++++
.../hive/execution/GlutenHiveSQLQuerySuite.scala | 62 ++++++++++++++++++++++
.../hive/execution/GlutenHiveSQLQuerySuite.scala | 62 ++++++++++++++++++++++
.../hive/execution/GlutenHiveSQLQuerySuite.scala | 62 ++++++++++++++++++++++
.../hive/execution/GlutenHiveSQLQuerySuite.scala | 62 ++++++++++++++++++++++
14 files changed, 356 insertions(+), 42 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 1415019681..09deec69e4 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 f81b78c71f..5f623abffc 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 0dc6f052e0..b742e3a5a0 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 d47d4c85ae..889e639464 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 4794437910..2a1fd3bf2b 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 36d029e8ff..b6dd5f9fa0 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 240033badb..109cf7b807 100644
--- a/cpp/velox/utils/ConfigExtractor.cc
+++ b/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)]
=
diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md
index 4787b07554..9ef1b3ba4b 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 the [...]
| 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 e4fafa3599..8ffb6345c3 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 e0ca4ae441..d7d6c57c5b 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 93b31a02e9..010e55c346 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 0938f0dd6c..5308ee7ede 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 0938f0dd6c..5308ee7ede 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 0938f0dd6c..5308ee7ede 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
