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 d52144e159 [GLUTEN-6887][VL] Pass Velox column mapping mode through
scan splits (#12884)
d52144e159 is described below
commit d52144e159fec87391f42c2581335d2e56c41c2e
Author: Rui Mo <[email protected]>
AuthorDate: Tue Sep 1 10:09:24 2026 +0100
[GLUTEN-6887][VL] Pass Velox column mapping mode through scan splits
(#12884)
---
.../delta/DeltaDeletionVectorScanInfoSuite.scala | 62 ++++++++++++-
.../delta/DeltaDeletionVectorScanInfoSuite.scala | 62 ++++++++++++-
.../gluten/execution/VeloxIcebergSuite.scala | 28 +++++-
.../backendsapi/velox/VeloxIteratorApi.scala | 24 +++++
.../org/apache/gluten/config/VeloxConfig.scala | 2 +-
cpp/velox/compute/VeloxPlanConverter.cc | 26 +++++-
cpp/velox/compute/WholeStageResultIterator.cc | 13 ++-
cpp/velox/compute/delta/DeltaSplit.cpp | 6 +-
cpp/velox/compute/delta/DeltaSplit.h | 3 +-
cpp/velox/compute/delta/tests/DeltaSplitTest.cpp | 5 +-
cpp/velox/config/VeloxConfig.h | 2 -
cpp/velox/substrait/SubstraitToVeloxPlan.h | 6 ++
.../tests/Substrait2VeloxPlanConversionTest.cc | 11 ++-
cpp/velox/tests/data/q6_first_stage_split.json | 11 ++-
cpp/velox/utils/ConfigExtractor.cc | 10 --
ep/build-velox/src/get-velox.sh | 4 +-
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 18 ++++
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 18 ++++
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 87 ++++++++++++++++-
.../gluten/delta/DeltaDeletionVectorScanInfo.scala | 87 ++++++++++++++++-
.../apache/gluten/delta/DeltaAddFileLookup.scala | 103 +++++++++++++++++++++
.../gluten/execution/DeltaScanTransformer.scala | 74 ++++++++++++++-
.../gluten/execution/IcebergScanTransformer.scala | 20 +++-
.../gluten/execution/PaimonScanTransformer.scala | 23 ++++-
.../gluten/substrait/rel/LocalFilesNode.java | 38 ++++++++
.../org/apache/gluten/config/GlutenConfig.scala | 30 ++----
.../hive/execution/GlutenHiveSQLQuerySuite.scala | 42 +++++++++
27 files changed, 751 insertions(+), 64 deletions(-)
diff --git
a/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala
b/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala
index 448bde4f84..633b19e32c 100644
---
a/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala
+++
b/backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.delta.{DeltaLog,
GlutenDeltaParquetFileFormat}
import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
import org.apache.spark.sql.delta.catalog.DeltaCatalog
import org.apache.spark.sql.delta.test.DeltaSQLTestUtils
+import org.apache.spark.sql.delta.util.DeltaFileOperations
import org.apache.spark.sql.execution.datasources.PartitionedFile
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
@@ -239,6 +240,64 @@ class DeltaDeletionVectorScanInfoSuite
DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile),
tablePath).get._2.head
}
+ test("normalize uses current AddFile when split DV metadata is stale") {
+ withTempDir {
+ tempDir =>
+ val tablePath = new Path(tempDir.getCanonicalPath, "spark%dir%prefix")
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+ .toDF("id", "value")
+ .coalesce(1)
+ .write
+ .format("delta")
+ .save(tablePath.toString)
+
+ spark.sql(
+ s"ALTER TABLE delta.`$tablePath` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
+ spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id = 4")
+ val staleFile = DeltaLog
+ .forTable(spark, tablePath)
+ .update()
+ .allFiles
+ .collect()
+ .find(_.deletionVector != null)
+ .get
+ val partitionedFile = partitionedFileWithMetadata(
+ tablePath.toString,
+ staleFile.path,
+ staleFile.size,
+ Map(
+ GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED ->
+ staleFile.deletionVector.serializeToBase64(),
+ GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE ->
"IF_CONTAINED",
+ "kept_key" -> "kept_value"
+ )
+ )
+
+ spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id = 3")
+ val currentFile = DeltaLog
+ .forTable(spark, tablePath)
+ .update()
+ .allFiles
+ .collect()
+ .find(_.path == staleFile.path)
+ .get
+ assert(currentFile.deletionVector.cardinality >
staleFile.deletionVector.cardinality)
+
+ val addFileLookup = DeltaDeletionVectorScanInfo
+ .buildAddFileLookup(tablePath, Seq(currentFile))
+ val result = DeltaDeletionVectorScanInfo.normalizeFromAddFiles(
+ Seq(partitionedFile),
+ tablePath,
+ addFileLookup)
+ assert(result.isDefined)
+ val (metadata, options) = result.get
+ assert(metadata.head.size() == 1)
+ assert(metadata.head.get("kept_key") == "kept_value")
+ assert(
+ options.head.deletionVectorCardinality ==
currentFile.deletionVector.cardinality)
+ }
+ }
+
override protected def partitionedFileWithMetadata(
tablePath: String,
relativeFilePath: String,
@@ -246,7 +305,8 @@ class DeltaDeletionVectorScanInfoSuite
metadata: Map[String, Object]): PartitionedFile = {
PartitionedFile(
partitionValues = InternalRow.empty,
- filePath = SparkPath.fromPath(new Path(tablePath, relativeFilePath)),
+ filePath = SparkPath.fromPath(
+ DeltaFileOperations.absolutePath(tablePath, relativeFilePath)),
start = 0L,
length = fileSize,
fileSize = fileSize,
diff --git
a/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala
b/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala
index 448bde4f84..633b19e32c 100644
---
a/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala
+++
b/backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.delta.{DeltaLog,
GlutenDeltaParquetFileFormat}
import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
import org.apache.spark.sql.delta.catalog.DeltaCatalog
import org.apache.spark.sql.delta.test.DeltaSQLTestUtils
+import org.apache.spark.sql.delta.util.DeltaFileOperations
import org.apache.spark.sql.execution.datasources.PartitionedFile
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
@@ -239,6 +240,64 @@ class DeltaDeletionVectorScanInfoSuite
DeltaDeletionVectorScanInfo.normalize(Seq(partitionedFile),
tablePath).get._2.head
}
+ test("normalize uses current AddFile when split DV metadata is stale") {
+ withTempDir {
+ tempDir =>
+ val tablePath = new Path(tempDir.getCanonicalPath, "spark%dir%prefix")
+ Seq((1, "a"), (2, "b"), (3, "c"), (4, "d"))
+ .toDF("id", "value")
+ .coalesce(1)
+ .write
+ .format("delta")
+ .save(tablePath.toString)
+
+ spark.sql(
+ s"ALTER TABLE delta.`$tablePath` SET TBLPROPERTIES
('delta.enableDeletionVectors' = true)")
+ spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id = 4")
+ val staleFile = DeltaLog
+ .forTable(spark, tablePath)
+ .update()
+ .allFiles
+ .collect()
+ .find(_.deletionVector != null)
+ .get
+ val partitionedFile = partitionedFileWithMetadata(
+ tablePath.toString,
+ staleFile.path,
+ staleFile.size,
+ Map(
+ GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_ID_ENCODED ->
+ staleFile.deletionVector.serializeToBase64(),
+ GlutenDeltaParquetFileFormat.FILE_ROW_INDEX_FILTER_TYPE ->
"IF_CONTAINED",
+ "kept_key" -> "kept_value"
+ )
+ )
+
+ spark.sql(s"DELETE FROM delta.`$tablePath` WHERE id = 3")
+ val currentFile = DeltaLog
+ .forTable(spark, tablePath)
+ .update()
+ .allFiles
+ .collect()
+ .find(_.path == staleFile.path)
+ .get
+ assert(currentFile.deletionVector.cardinality >
staleFile.deletionVector.cardinality)
+
+ val addFileLookup = DeltaDeletionVectorScanInfo
+ .buildAddFileLookup(tablePath, Seq(currentFile))
+ val result = DeltaDeletionVectorScanInfo.normalizeFromAddFiles(
+ Seq(partitionedFile),
+ tablePath,
+ addFileLookup)
+ assert(result.isDefined)
+ val (metadata, options) = result.get
+ assert(metadata.head.size() == 1)
+ assert(metadata.head.get("kept_key") == "kept_value")
+ assert(
+ options.head.deletionVectorCardinality ==
currentFile.deletionVector.cardinality)
+ }
+ }
+
override protected def partitionedFileWithMetadata(
tablePath: String,
relativeFilePath: String,
@@ -246,7 +305,8 @@ class DeltaDeletionVectorScanInfoSuite
metadata: Map[String, Object]): PartitionedFile = {
PartitionedFile(
partitionValues = InternalRow.empty,
- filePath = SparkPath.fromPath(new Path(tablePath, relativeFilePath)),
+ filePath = SparkPath.fromPath(
+ DeltaFileOperations.absolutePath(tablePath, relativeFilePath)),
start = 0L,
length = fileSize,
fileSize = fileSize,
diff --git
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
index 4e540c15af..c956029392 100644
---
a/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
+++
b/backends-velox/src-iceberg/test/scala/org/apache/gluten/execution/VeloxIcebergSuite.scala
@@ -16,7 +16,7 @@
*/
package org.apache.gluten.execution
-import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
import org.apache.spark.sql.Row
import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
@@ -27,6 +27,32 @@ import org.apache.iceberg.spark.source.SparkTable
import org.apache.iceberg.types.{Type, Types}
class VeloxIcebergSuite extends IcebergSuite {
+ test("iceberg parquet split uses name mapping for projected columns") {
+ withTable("iceberg_parquet_name_mapping") {
+ withSQLConf(VeloxConfig.PARQUET_USE_COLUMN_NAMES.key -> "false") {
+ spark.sql("""
+ |CREATE TABLE iceberg_parquet_name_mapping (
+ | id BIGINT,
+ | amount DECIMAL(12, 2),
+ | note STRING
+ |)
+ |USING iceberg
+ |TBLPROPERTIES ('write.format.default' = 'parquet')
+ |""".stripMargin)
+ spark.sql("""
+ |INSERT INTO iceberg_parquet_name_mapping
+ |VALUES (CAST(1 AS BIGINT), CAST(10.50 AS DECIMAL(12, 2)),
'a')
+ |""".stripMargin)
+
+ runQueryAndCompare("SELECT amount FROM iceberg_parquet_name_mapping") {
+ df =>
+ checkAnswer(df, Seq(Row(BigDecimal("10.50"))))
+ checkGlutenPlan[IcebergScanTransformer](df)
+ }
+ }
+ }
+ }
+
testWithMinSparkVersion("iceberg v3 initial default for an added column",
"3.4") {
withTable("iceberg_v3_initial_default") {
withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
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 5f623abffc..24ae3aa7ae 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
@@ -25,6 +25,7 @@ import org.apache.gluten.metrics.{IMetrics,
IteratorMetricsJniWrapper}
import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.substrait.plan.PlanNode
import org.apache.gluten.substrait.rel.{LocalFilesBuilder, LocalFilesNode,
SplitInfo}
+import org.apache.gluten.substrait.rel.LocalFilesNode.ColumnMappingMode
import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
import org.apache.gluten.vectorized._
@@ -35,6 +36,7 @@ import
org.apache.spark.sql.catalyst.catalog.ExternalCatalogUtils
import org.apache.spark.sql.catalyst.util.{DateFormatter, TimestampFormatter}
import org.apache.spark.sql.execution.datasources.{FilePartition,
PartitionedFile}
import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types._
import org.apache.spark.sql.utils.SparkInputMetricsUtil.InputMetricsWrapper
import org.apache.spark.sql.vectorized.ColumnarBatch
@@ -66,10 +68,32 @@ class VeloxIteratorApi extends IteratorApi with Logging {
) {
localFilesNode.setFileSchema(fileSchema)
}
+ columnMappingMode(fileFormat).foreach(localFilesNode.setColumnMappingMode)
localFilesNode
}
+ private def columnMappingMode(fileFormat: ReadFileFormat):
Option[ColumnMappingMode] = {
+ fileFormat match {
+ case ReadFileFormat.OrcReadFormat | ReadFileFormat.DwrfReadFormat =>
+ val defaultForcePosition =
+ org.apache.spark.SparkEnv.get.conf.get(
+ GlutenConfig.SPARK_ORC_FORCE_POSITIONAL_EVOLUTION,
+ "false")
+ val forcePosition =
+ SQLConf.get
+ .getConfString(GlutenConfig.SPARK_ORC_FORCE_POSITIONAL_EVOLUTION,
defaultForcePosition)
+ .toBoolean
+ Some(if (forcePosition) ColumnMappingMode.POSITION else
ColumnMappingMode.NAME)
+ case ReadFileFormat.ParquetReadFormat =>
+ Some(
+ if (VeloxConfig.get.parquetUseColumnNames) ColumnMappingMode.NAME
+ else ColumnMappingMode.POSITION)
+ case _ =>
+ None
+ }
+ }
+
override def genSplitInfo(
partitionIndex: Int,
partitions: Seq[Partition],
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 67c7192d95..d1d9e365f4 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
@@ -939,7 +939,7 @@ object VeloxConfig extends ConfigRegistry {
.createWithDefault(100)
val PARQUET_USE_COLUMN_NAMES =
- buildConf("spark.gluten.sql.columnar.backend.velox.parquetUseColumnNames")
+ buildConf(GlutenConfig.VELOX_PARQUET_USE_COLUMN_NAMES)
.doc("Maps table field names to file field names using names, not
indices for Parquet files.")
.booleanConf
.createWithDefault(true)
diff --git a/cpp/velox/compute/VeloxPlanConverter.cc
b/cpp/velox/compute/VeloxPlanConverter.cc
index 82c1f29036..2b5fb8bd9a 100644
--- a/cpp/velox/compute/VeloxPlanConverter.cc
+++ b/cpp/velox/compute/VeloxPlanConverter.cc
@@ -53,6 +53,10 @@ VeloxPlanConverter::VeloxPlanConverter(
}
namespace {
+// Keep this key in sync with the JVM-side constant
+// LocalFilesNode.COLUMN_MAPPING_MODE_METADATA_KEY.
+constexpr std::string_view kColumnMappingModeMetadataKey =
"__gluten.column_mapping_mode";
+
std::optional<std::string> unpackMetadataValue(const google::protobuf::Any&
value) {
google::protobuf::BytesValue bytesValue;
if (value.UnpackTo(&bytesValue)) {
@@ -79,8 +83,9 @@ std::optional<std::string> unpackMetadataValue(const
google::protobuf::Any& valu
return std::to_string(doubleValue.value());
}
- // Matches the string encoding the JVM side uses for booleans, which are
packed through
- // SubstraitUtil.convertJavaObjectToAny's toString fallback rather than as
BoolValue.
+ // Matches the string encoding the JVM side uses for booleans, which are
+ // packed through SubstraitUtil.convertJavaObjectToAny's toString fallback
+ // rather than as BoolValue.
google::protobuf::BoolValue boolValue;
if (value.UnpackTo(&boolValue)) {
return boolValue.value() ? "true" : "false";
@@ -89,6 +94,14 @@ std::optional<std::string> unpackMetadataValue(const
google::protobuf::Any& valu
return std::nullopt;
}
+std::optional<velox::dwio::common::ColumnMappingMode>
parseColumnMappingMode(const google::protobuf::Any& value) {
+ auto unpacked = unpackMetadataValue(value);
+ if (!unpacked.has_value()) {
+ return std::nullopt;
+ }
+ return
velox::dwio::common::ColumnMappingModeName::tryToColumnMappingMode(*unpacked);
+}
+
delta::DeltaRowIndexFilterType parseDeltaRowIndexFilterType(int filterType) {
switch (filterType) {
case 1:
@@ -164,6 +177,15 @@ std::shared_ptr<SplitInfo> parseScanSplitInfo(
metadataColumnMap[metadataColumn.key()] = metadataColumn.value();
}
for (const auto& otherMetadataColumn :
file.other_const_metadata_columns()) {
+ if (otherMetadataColumn.key() == kColumnMappingModeMetadataKey) {
+ auto mode = parseColumnMappingMode(otherMetadataColumn.value());
+ VELOX_CHECK(mode.has_value(), "Invalid column mapping mode metadata
for key {}", kColumnMappingModeMetadataKey);
+ if (splitInfo->columnMappingMode.has_value()) {
+ VELOX_CHECK_EQ(*splitInfo->columnMappingMode, *mode, "A single
SplitInfo cannot mix column mapping modes");
+ }
+ splitInfo->columnMappingMode = *mode;
+ continue;
+ }
if (auto unpackedValue =
unpackMetadataValue(otherMetadataColumn.value())) {
metadataColumnMap[otherMetadataColumn.key()] =
std::move(*unpackedValue);
}
diff --git a/cpp/velox/compute/WholeStageResultIterator.cc
b/cpp/velox/compute/WholeStageResultIterator.cc
index 409f288c17..fcfb55bfc1 100644
--- a/cpp/velox/compute/WholeStageResultIterator.cc
+++ b/cpp/velox/compute/WholeStageResultIterator.cc
@@ -188,7 +188,10 @@ WholeStageResultIterator::WholeStageResultIterator(
true,
deleteFiles,
metadataColumn,
- properties[idx]);
+ properties[idx],
+ /*dataSequenceNumber=*/0,
+ /*identityPartitionKeys=*/std::unordered_map<int32_t,
std::optional<std::string>>{},
+ scanInfo->columnMappingMode);
} else if (isDeltaScan) {
std::unordered_map<std::string, std::string>
customSplitInfo{{"table_format", kDeltaTableFormat}};
std::optional<gluten::delta::DeltaDeletionVectorDescriptor>
deletionVector = std::nullopt;
@@ -215,7 +218,8 @@ WholeStageResultIterator::WholeStageResultIterator(
std::nullopt,
rowIndexFilterType,
metadataColumn,
- properties[idx]);
+ properties[idx],
+ scanInfo->columnMappingMode);
} else {
auto connectorId = connectorIds_.hive;
#ifdef GLUTEN_ENABLE_GPU
@@ -238,7 +242,10 @@ WholeStageResultIterator::WholeStageResultIterator(
0,
true,
metadataColumn,
- properties[idx]);
+ properties[idx],
+ std::nullopt,
+ std::nullopt,
+ scanInfo->columnMappingMode);
}
connectorSplits.emplace_back(split);
}
diff --git a/cpp/velox/compute/delta/DeltaSplit.cpp
b/cpp/velox/compute/delta/DeltaSplit.cpp
index b5b22a3ab5..bb72a82f64 100644
--- a/cpp/velox/compute/delta/DeltaSplit.cpp
+++ b/cpp/velox/compute/delta/DeltaSplit.cpp
@@ -35,7 +35,8 @@ HiveDeltaSplit::HiveDeltaSplit(
std::optional<DeltaFileStatistics> statistics,
DeltaRowIndexFilterType filterType,
const std::unordered_map<std::string, std::string>& infoColumns,
- std::optional<FileProperties> fileProperties)
+ std::optional<FileProperties> fileProperties,
+ std::optional<dwio::common::ColumnMappingMode> columnMappingMode)
: HiveConnectorSplit(
connectorId,
filePath,
@@ -52,7 +53,8 @@ HiveDeltaSplit::HiveDeltaSplit(
infoColumns,
fileProperties,
std::nullopt,
- std::nullopt),
+ std::nullopt,
+ columnMappingMode),
deletionVector(std::move(deletionVector)),
statistics(std::move(statistics)),
filterType(filterType) {}
diff --git a/cpp/velox/compute/delta/DeltaSplit.h
b/cpp/velox/compute/delta/DeltaSplit.h
index 3f9d9dd347..a99349ee47 100644
--- a/cpp/velox/compute/delta/DeltaSplit.h
+++ b/cpp/velox/compute/delta/DeltaSplit.h
@@ -101,7 +101,8 @@ struct HiveDeltaSplit : public
connector::hive::HiveConnectorSplit {
std::optional<DeltaFileStatistics> statistics = std::nullopt,
DeltaRowIndexFilterType filterType = DeltaRowIndexFilterType::kKeepAll,
const std::unordered_map<std::string, std::string>& infoColumns = {},
- std::optional<FileProperties> fileProperties = std::nullopt);
+ std::optional<FileProperties> fileProperties = std::nullopt,
+ std::optional<dwio::common::ColumnMappingMode> columnMappingMode =
std::nullopt);
};
} // namespace gluten::delta
diff --git a/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp
b/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp
index 13482ccec3..4687a5771e 100644
--- a/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp
+++ b/cpp/velox/compute/delta/tests/DeltaSplitTest.cpp
@@ -56,13 +56,16 @@ TEST(DeltaSplitTest, SplitCarriesDeletionVectorDescriptor) {
std::nullopt,
DeltaRowIndexFilterType::kIfContained,
std::unordered_map<std::string, std::string>{},
- std::nullopt);
+ std::nullopt,
+ facebook::velox::dwio::common::ColumnMappingMode::kName);
ASSERT_TRUE(split->deletionVector.has_value());
EXPECT_EQ(split->deletionVector->cardinality, 2);
ASSERT_TRUE(split->deletionVector->serializedPayloadView.has_value());
EXPECT_EQ(split->deletionVector->serializedPayloadView->size,
payload.size());
EXPECT_EQ(split->filterType, DeltaRowIndexFilterType::kIfContained);
+ ASSERT_TRUE(split->columnMappingMode.has_value());
+ EXPECT_EQ(split->columnMappingMode.value(),
facebook::velox::dwio::common::ColumnMappingMode::kName);
}
TEST(DeltaSplitTest, LogicalRowCountSubtractsDeletionVectorCardinality) {
diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h
index f9351560cf..8a37e36dcb 100644
--- a/cpp/velox/config/VeloxConfig.h
+++ b/cpp/velox/config/VeloxConfig.h
@@ -196,8 +196,6 @@ 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 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";
// write files
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.h
b/cpp/velox/substrait/SubstraitToVeloxPlan.h
index b0cf76fff3..2464881052 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.h
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.h
@@ -17,6 +17,8 @@
#pragma once
+#include <optional>
+
#include "SubstraitToVeloxExpr.h"
#include "TypeUtils.h"
#include "compute/VeloxConnectorIds.h"
@@ -68,6 +70,10 @@ struct SplitInfo {
/// The schema of the table being scanned.
RowTypePtr tableSchema;
+ /// Optional scan-level column mapping mode. Gluten may set this differently
+ /// for different scans or partitions in the same Velox QueryCtx.
+ std::optional<dwio::common::ColumnMappingMode> columnMappingMode;
+
/// Make SplitInfo polymorphic
virtual ~SplitInfo() = default;
diff --git a/cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc
b/cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc
index d002f34a14..aa02a3f4e8 100644
--- a/cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc
+++ b/cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc
@@ -59,11 +59,12 @@ class Substrait2VeloxPlanConversionTest : public
exec::test::HiveConnectorTestBa
auto path = fmt::format("{}{}", tmpDir_->getPath(), paths[i]);
auto start = starts[i];
auto length = lengths[i];
- auto split = facebook::velox::exec::test::HiveConnectorSplitBuilder(path)
- .fileFormat(fileFormat)
- .start(start)
- .length(length)
- .build();
+ facebook::velox::exec::test::HiveConnectorSplitBuilder
splitBuilder(path);
+ splitBuilder.fileFormat(fileFormat).start(start).length(length);
+ if (splitInfo->columnMappingMode.has_value()) {
+ splitBuilder.columnMappingMode(*splitInfo->columnMappingMode);
+ }
+ auto split = splitBuilder.build();
splits.emplace_back(split);
}
return splits;
diff --git a/cpp/velox/tests/data/q6_first_stage_split.json
b/cpp/velox/tests/data/q6_first_stage_split.json
index 7b02e4dc01..2ebcd63e93 100644
--- a/cpp/velox/tests/data/q6_first_stage_split.json
+++ b/cpp/velox/tests/data/q6_first_stage_split.json
@@ -5,7 +5,16 @@
"start": "0",
"length": "3719",
"uri_file": "/mock_lineitem.dwrf",
+ "other_const_metadata_columns": [
+ {
+ "key": "__gluten.column_mapping_mode",
+ "value": {
+ "@type":
"type.googleapis.com/google.protobuf.StringValue",
+ "value": "NAME"
+ }
+ }
+ ],
"dwrf": {}
}
]
-}
\ No newline at end of file
+}
diff --git a/cpp/velox/utils/ConfigExtractor.cc
b/cpp/velox/utils/ConfigExtractor.cc
index d38745416b..ed2850a811 100644
--- a/cpp/velox/utils/ConfigExtractor.cc
+++ b/cpp/velox/utils/ConfigExtractor.cc
@@ -269,18 +269,8 @@ std::shared_ptr<facebook::velox::config::ConfigBase>
createHiveConnectorSessionC
conf->get<std::string>(kParquetMaxTargetFileSize, "0B"); // 0 means no
limit on target file size
configs[facebook::velox::connector::hive::HiveConfig::kIgnoreMissingFilesSession]
=
conf->get<bool>(kIgnoreMissingFiles, false) ? "true" : "false";
-
configs[parquetSessionProperty(facebook::velox::parquet::ParquetConfig::kUseColumnNamesSession)]
=
- 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[orcSessionProperty(facebook::velox::dwrf::Config::kOrcUseColumnNamesSession)]
=
- 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/ep/build-velox/src/get-velox.sh b/ep/build-velox/src/get-velox.sh
index 7c8ec8cba3..b7f9f0558c 100755
--- a/ep/build-velox/src/get-velox.sh
+++ b/ep/build-velox/src/get-velox.sh
@@ -18,8 +18,8 @@ set -exu
CURRENT_DIR=$(cd "$(dirname "$BASH_SOURCE")"; pwd)
VELOX_REPO=https://github.com/IBM/velox.git
-VELOX_BRANCH=dft-2026_08_26
-VELOX_ENHANCED_BRANCH=ibm-2026_08_26
+VELOX_BRANCH=dft-2026_08_26_fix
+VELOX_ENHANCED_BRANCH=dft-2026_08_26_fix
VELOX_HOME=""
RUN_SETUP_SCRIPT=ON
ENABLE_ENHANCED_FEATURES=OFF
diff --git
a/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
b/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
index 084f3d3319..1fc0f2c025 100644
---
a/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
+++
b/gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
@@ -18,6 +18,7 @@ package org.apache.gluten.delta
import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
+import org.apache.spark.sql.delta.actions.AddFile
import org.apache.spark.sql.execution.datasources.PartitionedFile
import org.apache.hadoop.fs.Path
@@ -38,4 +39,21 @@ object DeltaDeletionVectorScanInfo {
tablePath: Path,
readMetrics: Option[DeletionVectorReadMetrics])
: Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
+
+ private[gluten] def buildAddFileLookup(
+ tablePath: Path,
+ addFiles: Seq[AddFile]): DeltaAddFileLookup = DeltaAddFileLookup.empty
+
+ private[gluten] def normalizeFromAddFiles(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ addFileLookup: DeltaAddFileLookup)
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
+
+ private[gluten] def normalizeFromAddFiles(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ addFileLookup: DeltaAddFileLookup,
+ readMetrics: Option[DeletionVectorReadMetrics])
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
}
diff --git
a/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
b/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
index 084f3d3319..1fc0f2c025 100644
---
a/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
+++
b/gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
@@ -18,6 +18,7 @@ package org.apache.gluten.delta
import org.apache.gluten.substrait.rel.DeltaLocalFilesNode.DeltaFileReadOptions
+import org.apache.spark.sql.delta.actions.AddFile
import org.apache.spark.sql.execution.datasources.PartitionedFile
import org.apache.hadoop.fs.Path
@@ -38,4 +39,21 @@ object DeltaDeletionVectorScanInfo {
tablePath: Path,
readMetrics: Option[DeletionVectorReadMetrics])
: Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
+
+ private[gluten] def buildAddFileLookup(
+ tablePath: Path,
+ addFiles: Seq[AddFile]): DeltaAddFileLookup = DeltaAddFileLookup.empty
+
+ private[gluten] def normalizeFromAddFiles(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ addFileLookup: DeltaAddFileLookup)
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
+
+ private[gluten] def normalizeFromAddFiles(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ addFileLookup: DeltaAddFileLookup,
+ readMetrics: Option[DeletionVectorReadMetrics])
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
}
diff --git
a/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
b/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
index 71a2cd265a..9f717b25e8 100644
---
a/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
+++
b/gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
@@ -22,7 +22,7 @@ import
org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeletionVectorPayloa
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.delta.DeltaParquetFileFormat
-import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
+import org.apache.spark.sql.delta.actions.{AddFile, DeletionVectorDescriptor}
import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArrayFormat,
StoredBitmap}
import org.apache.spark.sql.delta.storage.dv.{DeletionVectorStore,
HadoopFileSystemDVStore}
import org.apache.spark.sql.execution.datasources.PartitionedFile
@@ -104,6 +104,60 @@ object DeltaDeletionVectorScanInfo {
}
}
+ /**
+ * Materializes normal-table DV options from the AddFiles selected by
PreparedDeltaFileIndex.
+ * These are authoritative when a data file's DV is replaced by repeated
DML; the corresponding
+ * PartitionedFile may still contain the previous descriptor in its constant
metadata.
+ */
+ private[gluten] def buildAddFileLookup(
+ tablePath: Path,
+ addFiles: Seq[AddFile]): DeltaAddFileLookup = {
+ DeltaAddFileLookup(tablePath, addFiles, addFiles.exists(_.deletionVector
!= null))
+ }
+
+ private[gluten] def normalizeFromAddFiles(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ addFileLookup: DeltaAddFileLookup)
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
+ normalizeFromAddFiles(partitionFiles, tablePath, addFileLookup, None)
+ }
+
+ private[gluten] def normalizeFromAddFiles(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ addFileLookup: DeltaAddFileLookup,
+ readMetrics: Option[DeletionVectorReadMetrics])
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
+ if (partitionFiles.isEmpty) {
+ return None
+ }
+ val spark = activeSparkSession
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val serializableHadoopConf = new SerializableConfiguration(hadoopConf)
+ val hadDvMetadata = partitionFiles.exists {
+ file =>
+ val metadata = otherMetadataColumns(file)
+ metadata.contains(RowIndexFilterIdEncoded) ||
metadata.contains(RowIndexFilterTypeKey)
+ }
+ if (!hadDvMetadata && !addFileLookup.hasDeletionVector) {
+ return None
+ }
+ val scanInfos = partitionFiles.map {
+ file =>
+ val addFile = addFileLookup.find(file)
+ extract(file, hadoopConf, serializableHadoopConf, tablePath, addFile,
readMetrics)
+ }
+ if (hadDvMetadata ||
scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) {
+ Some(
+ (
+ scanInfos.map(_.normalizedOtherMetadataColumns.asJava),
+ scanInfos.map(info =>
toDeltaFileReadOptions(info.deletionVectorInfo))))
+ } else {
+ None
+ }
+ }
+
/** Public entry point for extracting DV info from a single file (used by
tests). */
def extract(
spark: SparkSession,
@@ -131,6 +185,37 @@ object DeltaDeletionVectorScanInfo {
PartitionFileScanInfo(normalizedMetadata, dvInfo)
}
+ private def extract(
+ file: PartitionedFile,
+ hadoopConf: Configuration,
+ serializableHadoopConf: SerializableConfiguration,
+ tablePath: Path,
+ addFile: AddFile,
+ readMetrics: Option[DeletionVectorReadMetrics]): PartitionFileScanInfo =
{
+ val metadata = otherMetadataColumns(file)
+ val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded,
RowIndexFilterTypeKey)
+ val dvInfo = Option(addFile.deletionVector) match {
+ case Some(descriptor) =>
+ DeletionVectorInfo(
+ true,
+ IF_CONTAINED,
+ descriptor.cardinality,
+ deletionVectorPayload(
+ hadoopConf,
+ serializableHadoopConf,
+ tablePath,
+ descriptor,
+ readMetrics))
+ case None =>
+ DeletionVectorInfo(
+ false,
+ KEEP_ALL,
+ 0L,
+ new InMemoryDeletionVectorPayload(Array.emptyByteArray))
+ }
+ PartitionFileScanInfo(normalizedMetadata, dvInfo)
+ }
+
private def toDeltaFileReadOptions(dvInfo: DeletionVectorInfo):
DeltaFileReadOptions = {
new DeltaFileReadOptions(
toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType),
diff --git
a/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
b/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
index e27d10774f..cf5dd1a3ac 100644
---
a/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
+++
b/gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala
@@ -22,7 +22,7 @@ import
org.apache.gluten.substrait.rel.DeltaLocalFilesNode.{DeletionVectorPayloa
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.delta.DeltaParquetFileFormat
-import org.apache.spark.sql.delta.actions.DeletionVectorDescriptor
+import org.apache.spark.sql.delta.actions.{AddFile, DeletionVectorDescriptor}
import org.apache.spark.sql.delta.deletionvectors.{RoaringBitmapArrayFormat,
StoredBitmap}
import org.apache.spark.sql.delta.storage.dv.{DeletionVectorStore,
HadoopFileSystemDVStore}
import org.apache.spark.sql.execution.datasources.PartitionedFile
@@ -105,6 +105,60 @@ object DeltaDeletionVectorScanInfo {
}
}
+ /**
+ * Materializes normal-table DV options from the AddFiles selected by
PreparedDeltaFileIndex.
+ * These are authoritative when a data file's DV is replaced by repeated
DML; the corresponding
+ * PartitionedFile may still contain the previous descriptor in its constant
metadata.
+ */
+ private[gluten] def buildAddFileLookup(
+ tablePath: Path,
+ addFiles: Seq[AddFile]): DeltaAddFileLookup = {
+ DeltaAddFileLookup(tablePath, addFiles, addFiles.exists(_.deletionVector
!= null))
+ }
+
+ private[gluten] def normalizeFromAddFiles(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ addFileLookup: DeltaAddFileLookup)
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
+ normalizeFromAddFiles(partitionFiles, tablePath, addFileLookup, None)
+ }
+
+ private[gluten] def normalizeFromAddFiles(
+ partitionFiles: Seq[PartitionedFile],
+ tablePath: Path,
+ addFileLookup: DeltaAddFileLookup,
+ readMetrics: Option[DeletionVectorReadMetrics])
+ : Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = {
+ if (partitionFiles.isEmpty) {
+ return None
+ }
+ val spark = activeSparkSession
+ val hadoopConf = spark.sessionState.newHadoopConf()
+ val serializableHadoopConf = new SerializableConfiguration(hadoopConf)
+ val hadDvMetadata = partitionFiles.exists {
+ file =>
+ val metadata = otherMetadataColumns(file)
+ metadata.contains(RowIndexFilterIdEncoded) ||
metadata.contains(RowIndexFilterTypeKey)
+ }
+ if (!hadDvMetadata && !addFileLookup.hasDeletionVector) {
+ return None
+ }
+ val scanInfos = partitionFiles.map {
+ file =>
+ val addFile = addFileLookup.find(file)
+ extract(file, hadoopConf, serializableHadoopConf, tablePath, addFile,
readMetrics)
+ }
+ if (hadDvMetadata ||
scanInfos.exists(_.deletionVectorInfo.hasDeletionVector)) {
+ Some(
+ (
+ scanInfos.map(_.normalizedOtherMetadataColumns.asJava),
+ scanInfos.map(info =>
toDeltaFileReadOptions(info.deletionVectorInfo))))
+ } else {
+ None
+ }
+ }
+
/** Public entry point for extracting DV info from a single file (used by
tests). */
def extract(
spark: SparkSession,
@@ -132,6 +186,37 @@ object DeltaDeletionVectorScanInfo {
PartitionFileScanInfo(normalizedMetadata, dvInfo)
}
+ private def extract(
+ file: PartitionedFile,
+ hadoopConf: Configuration,
+ serializableHadoopConf: SerializableConfiguration,
+ tablePath: Path,
+ addFile: AddFile,
+ readMetrics: Option[DeletionVectorReadMetrics]): PartitionFileScanInfo =
{
+ val metadata = otherMetadataColumns(file)
+ val normalizedMetadata = metadata -- Seq(RowIndexFilterIdEncoded,
RowIndexFilterTypeKey)
+ val dvInfo = Option(addFile.deletionVector) match {
+ case Some(descriptor) =>
+ DeletionVectorInfo(
+ true,
+ IF_CONTAINED,
+ descriptor.cardinality,
+ deletionVectorPayload(
+ hadoopConf,
+ serializableHadoopConf,
+ tablePath,
+ descriptor,
+ readMetrics))
+ case None =>
+ DeletionVectorInfo(
+ false,
+ KEEP_ALL,
+ 0L,
+ new InMemoryDeletionVectorPayload(Array.emptyByteArray))
+ }
+ PartitionFileScanInfo(normalizedMetadata, dvInfo)
+ }
+
private def toDeltaFileReadOptions(dvInfo: DeletionVectorInfo):
DeltaFileReadOptions = {
new DeltaFileReadOptions(
toSubstraitRowIndexFilterType(dvInfo.rowIndexFilterType),
diff --git
a/gluten-delta/src/main/scala/org/apache/gluten/delta/DeltaAddFileLookup.scala
b/gluten-delta/src/main/scala/org/apache/gluten/delta/DeltaAddFileLookup.scala
new file mode 100644
index 0000000000..8b2ff853f3
--- /dev/null
+++
b/gluten-delta/src/main/scala/org/apache/gluten/delta/DeltaAddFileLookup.scala
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.delta
+
+import org.apache.spark.sql.delta.actions.AddFile
+import org.apache.spark.sql.delta.util.DeltaFileOperations
+import org.apache.spark.sql.execution.datasources.PartitionedFile
+
+import org.apache.hadoop.fs.Path
+
+import scala.collection.mutable
+
+/** Driver-side AddFile index shared by all FilePartitions in one prepared
Delta scan. */
+final private[gluten] class DeltaAddFileLookup private (
+ addFiles: IndexedSeq[AddFile],
+ absolutePaths: IndexedSeq[Path],
+ candidateIndexes: Map[(Option[String], String), Vector[Int]],
+ val hasDeletionVector: Boolean) {
+
+ def find(file: PartitionedFile): AddFile = {
+ val partitionedFilePath = new Path(file.filePath.toString)
+ val candidates = mutable.BitSet.empty
+ DeltaAddFileLookup.pathVariants(partitionedFilePath).foreach {
+ key => candidateIndexes.get(key).foreach(indexes => candidates ++=
indexes)
+ }
+
+ // BitSet iteration preserves the original AddFile order used by the
previous Seq.find lookup.
+ candidates.iterator
+ .find(index => DeltaAddFileLookup.samePath(partitionedFilePath,
absolutePaths(index)))
+ .map(addFiles.apply)
+ .getOrElse {
+ throw new IllegalStateException(
+ s"Unable to find Delta AddFile metadata for split ${file.filePath}")
+ }
+ }
+}
+
+private[gluten] object DeltaAddFileLookup {
+ val empty: DeltaAddFileLookup =
+ new DeltaAddFileLookup(Vector.empty, Vector.empty, Map.empty,
hasDeletionVector = false)
+
+ def apply(
+ tablePath: Path,
+ addFiles: Seq[AddFile],
+ hasDeletionVector: Boolean): DeltaAddFileLookup = {
+ val indexedAddFiles = addFiles.toIndexedSeq
+ val absolutePaths = indexedAddFiles.map {
+ addFile => DeltaFileOperations.absolutePath(tablePath.toString,
addFile.path)
+ }
+ val candidateIndexes = mutable.HashMap.empty[
+ (Option[String], String),
+ mutable.ArrayBuffer[Int]]
+
+ absolutePaths.zipWithIndex.foreach {
+ case (absolutePath, index) =>
+ pathVariants(absolutePath).foreach {
+ key => candidateIndexes.getOrElseUpdate(key,
mutable.ArrayBuffer.empty) += index
+ }
+ }
+
+ new DeltaAddFileLookup(
+ indexedAddFiles,
+ absolutePaths,
+ candidateIndexes.iterator.map { case (key, indexes) => key ->
indexes.toVector }.toMap,
+ hasDeletionVector)
+ }
+
+ private def samePath(left: Path, right: Path): Boolean = {
+ pathVariants(left).intersect(pathVariants(right)).nonEmpty
+ }
+
+ private def pathVariants(path: Path): Set[(Option[String], String)] = {
+ val uri = path.toUri.normalize()
+ val authority = Option(uri.getAuthority)
+ Seq(uri.getRawPath, uri.getPath)
+ .filter(_ != null)
+ .flatMap(percentVariants)
+ .map(pathValue => authority -> pathValue)
+ .toSet
+ }
+
+ // SparkPath and DeltaFileOperations can expose literal '%' characters at
different URI escaping
+ // levels. Compare a bounded set of full-path variants while retaining the
URI authority.
+ private def percentVariants(path: String): Set[String] = {
+ (0 until 4).foldLeft(Set(path)) {
+ (variants, _) => variants ++ variants.map(_.replace("%25", "%"))
+ }
+ }
+}
diff --git
a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
b/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
index 3d80c582d7..fe26accb71 100644
---
a/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
+++
b/gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala
@@ -19,6 +19,7 @@ package org.apache.gluten.execution
import org.apache.gluten.delta.{DeletionVectorReadMetrics,
DeltaDeletionVectorScanInfo}
import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.substrait.rel.{DeltaLocalFilesBuilder,
LocalFilesNode, SplitInfo}
+import org.apache.gluten.substrait.rel.LocalFilesNode.ColumnMappingMode
import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
import org.apache.spark.Partition
@@ -26,8 +27,9 @@ import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, Expression}
import org.apache.spark.sql.catalyst.plans.QueryPlan
import org.apache.spark.sql.connector.read.streaming.SparkDataStream
-import org.apache.spark.sql.delta.{DeltaParquetFileFormat, NoMapping}
+import org.apache.spark.sql.delta.{DeltaParquetFileFormat, NameMapping,
NoMapping}
import org.apache.spark.sql.delta.files.{CdcAddFileIndex, TahoeFileIndex,
TahoeRemoveFileIndex}
+import org.apache.spark.sql.delta.stats.PreparedDeltaFileIndex
import org.apache.spark.sql.execution.FileSourceScanExec
import org.apache.spark.sql.execution.datasources.{FilePartition,
HadoopFsRelation}
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
@@ -142,11 +144,63 @@ case class DeltaScanTransformer(
override def getSplitInfosFromPartitions(
partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = {
val splitInfos = super.getSplitInfosFromPartitions(partitions)
- // Deletion vectors only exist on Delta tables read through a
TahoeFileIndex (which also covers
- // PreparedDeltaFileIndex). Its `path` is the authoritative table root and
is used to resolve
- // per-file DV locations. Any other location cannot carry Delta DV
metadata, so the generic
- // split representation is returned unchanged.
+ // Keep Delta's split decoration narrow. The generic Parquet path has
already attached the
+ // session-derived split mapping mode and only attaches file schema when
position mapping
+ // needs it. Delta name column mapping is the one case that must force
name mapping regardless
+ // of the generic Parquet setting because Gluten rewrites the scan schema
to physical names.
+ splitInfos.foreach {
+ case localFiles: LocalFilesNode =>
+ deltaColumnMappingMode.foreach {
+ mode =>
+ localFiles.clearFileSchema()
+ localFiles.setColumnMappingMode(mode)
+ }
+ case _ =>
+ }
+ // PreparedDeltaFileIndex contains the exact AddFiles selected for this
scan. Use these as the
+ // source of truth because PartitionedFile metadata can retain an older DV
descriptor after
+ // repeated DML updates the same data file.
relation.location match {
+ case prepared: PreparedDeltaFileIndex =>
+ val tableRootPath = prepared.path
+ val lookupStartedAt = System.nanoTime()
+ val addFileLookup =
+ try {
+ DeltaDeletionVectorScanInfo
+ .buildAddFileLookup(tableRootPath, prepared.preparedScan.files)
+ } finally {
+ metrics("dvDescriptorPreparationTime").add(System.nanoTime() -
lookupStartedAt)
+ }
+ splitInfos.zip(partitions).map {
+ case (localFiles: LocalFilesNode, (filePartition: FilePartition, _))
=>
+ val startedAt = System.nanoTime()
+ val normalized =
+ try {
+ DeltaDeletionVectorScanInfo
+ .normalizeFromAddFiles(
+ filePartition.files.toSeq,
+ tableRootPath,
+ addFileLookup,
+ Some(deletionVectorReadMetrics))
+ } finally {
+ metrics("dvDescriptorPreparationTime").add(System.nanoTime() -
startedAt)
+ }
+ normalized
+ .map {
+ case (otherMetadataColumns, deltaReadOptions) =>
+ metrics("dvDescriptorCount")
+ .add(deltaReadOptions.count(_.hasDeletionVector()).toLong)
+ DeltaLocalFilesBuilder.makeDeltaLocalFiles(
+ localFiles,
+ otherMetadataColumns.asJava,
+ deltaReadOptions.asJava): SplitInfo
+ }
+ .getOrElse(localFiles)
+ case (splitInfo, _) => splitInfo
+ }
+ // Other Tahoe indexes, such as CDF indexes, encode the row-index filter
type and DV
+ // descriptor in PartitionedFile metadata. Keep using that metadata for
these specialized
+ // scans because their semantics are not necessarily IF_CONTAINED.
case tahoe: TahoeFileIndex =>
val tableRootPath = tahoe.path
splitInfos.zip(partitions).map {
@@ -179,6 +233,16 @@ case class DeltaScanTransformer(
}
}
+ private def deltaColumnMappingMode: Option[ColumnMappingMode] =
relation.fileFormat match {
+ case d: DeltaParquetFileFormat =>
+ d.columnMappingMode match {
+ case NameMapping => Some(ColumnMappingMode.NAME)
+ // Preserves the previous Spark fallback behavior for IdMapping.
+ case _ => None
+ }
+ case _ => None
+ }
+
override def doCanonicalize(): DeltaScanTransformer = {
DeltaScanTransformer(
relation,
diff --git
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
index 16018e086d..911a98680b 100644
---
a/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
+++
b/gluten-iceberg/src/main/scala/org/apache/gluten/execution/IcebergScanTransformer.scala
@@ -21,6 +21,7 @@ import org.apache.gluten.exception.GlutenNotSupportException
import
org.apache.gluten.execution.IcebergScanTransformer.{containsMetadataColumn,
containsUuidOrFixedType}
import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.substrait.rel.{LocalFilesNode, SplitInfo}
+import org.apache.gluten.substrait.rel.LocalFilesNode.ColumnMappingMode
import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
import org.apache.spark.Partition
@@ -217,11 +218,15 @@ case class IcebergScanTransformer(
override def getSplitInfosFromPartitions(
partitions: Seq[(Partition, ReadFileFormat)]): Seq[SplitInfo] = {
val metadataColumnNames = getMetadataColumns().map(_.name)
- partitions.map { case (partition, _) => partitionToSplitInfo(partition,
metadataColumnNames) }
+ partitions.map {
+ case (partition, readFileFormat) =>
+ partitionToSplitInfo(partition, readFileFormat, metadataColumnNames)
+ }
}
private def partitionToSplitInfo(
partition: Partition,
+ readFileFormat: ReadFileFormat,
metadataColumnNames: Seq[String]): SplitInfo = {
val splitInfo = partition match {
case p: SparkDataSourceRDDPartition =>
@@ -233,10 +238,21 @@ case class IcebergScanTransformer(
icebergInitialDefaults)
case _ => throw new GlutenNotSupportException()
}
- numSplits.add(splitInfo.asInstanceOf[LocalFilesNode].getPaths.size())
+ val localFiles = splitInfo.asInstanceOf[LocalFilesNode]
+
icebergColumnMappingMode(readFileFormat).foreach(localFiles.setColumnMappingMode)
+ numSplits.add(localFiles.getPaths.size())
splitInfo
}
+ private def icebergColumnMappingMode(fileFormat: ReadFileFormat):
Option[ColumnMappingMode] = {
+ fileFormat match {
+ case ReadFileFormat.ParquetReadFormat | ReadFileFormat.OrcReadFormat =>
+ Some(ColumnMappingMode.NAME)
+ case _ =>
+ None
+ }
+ }
+
override def doCanonicalize(): IcebergScanTransformer = {
this.copy(
output = output.map(QueryPlan.normalizeExpressions(_, output)),
diff --git
a/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonScanTransformer.scala
b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonScanTransformer.scala
index 12bccb1568..998ec6e084 100644
---
a/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonScanTransformer.scala
+++
b/gluten-paimon/src-paimon/main/scala/org/apache/gluten/execution/PaimonScanTransformer.scala
@@ -19,6 +19,7 @@ package org.apache.gluten.execution
import org.apache.gluten.exception.GlutenNotSupportException
import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.gluten.substrait.rel.{PaimonLocalFilesBuilder, SplitInfo}
+import org.apache.gluten.substrait.rel.LocalFilesNode.ColumnMappingMode
import org.apache.gluten.substrait.rel.LocalFilesNode.ReadFileFormat
import org.apache.spark.Partition
@@ -83,7 +84,13 @@ case class PaimonScanTransformer(
throw new GlutenNotSupportException("Only support PaimonScan.")
}
- override def getDataSchema: StructType = new StructType()
+ override def getDataSchema: StructType = scan match {
+ case paimonScan: PaimonScan =>
+ val partitionKeys = paimonScan.table.partitionKeys()
+ StructType(scan.readSchema().filterNot(field =>
partitionKeys.contains(field.name)))
+ case _ =>
+ throw new GlutenNotSupportException("Only support PaimonScan.")
+ }
override def withNewPushdownFilters(filters: Seq[Expression]):
PaimonScanTransformer = {
this.copy(pushDownFilters = Some(filters))
@@ -165,7 +172,7 @@ case class PaimonScanTransformer(
throw new GlutenNotSupportException(s"Unsupported input partition
type: $o")
}
- PaimonLocalFilesBuilder.makePaimonLocalFiles(
+ val localFiles = PaimonLocalFilesBuilder.makePaimonLocalFiles(
p.index,
paths.asJava,
starts.asJava,
@@ -178,10 +185,22 @@ case class PaimonScanTransformer(
.asJava,
new JHashMap[String, String]()
)
+ localFiles.setFileSchema(getDataSchema)
+
paimonColumnMappingMode(fileFormat).foreach(localFiles.setColumnMappingMode)
+ localFiles
case _ => throw new GlutenNotSupportException()
}
}
+ private def paimonColumnMappingMode(fileFormat: ReadFileFormat):
Option[ColumnMappingMode] = {
+ fileFormat match {
+ case ReadFileFormat.ParquetReadFormat | ReadFileFormat.OrcReadFormat =>
+ Some(ColumnMappingMode.NAME)
+ case _ =>
+ None
+ }
+ }
+
override def doCanonicalize(): PaimonScanTransformer = {
this.copy(
output = output.map(QueryPlan.normalizeExpressions(_, output)),
diff --git
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java
index dfea5dd753..5070a88581 100644
---
a/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java
+++
b/gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/LocalFilesNode.java
@@ -31,6 +31,8 @@ import java.util.List;
import java.util.Map;
public class LocalFilesNode implements SplitInfo {
+ public static final String COLUMN_MAPPING_MODE_METADATA_KEY =
"__gluten.column_mapping_mode";
+
private final Integer index;
private final List<String> paths = new ArrayList<>();
private final List<Long> starts = new ArrayList<>();
@@ -55,10 +57,29 @@ public class LocalFilesNode implements SplitInfo {
UnknownFormat()
}
+ // This is fully aligned with mapping modes defined in Velox:
velox/dwio/common/Options.cpp.
+ public enum ColumnMappingMode {
+ POSITION("POSITION"),
+ NAME("NAME"),
+ PARQUET_FIELD_ID("PARQUET_FIELD_ID"),
+ FIELD_ID("FIELD_ID");
+
+ private final String nativeName;
+
+ ColumnMappingMode(String nativeName) {
+ this.nativeName = nativeName;
+ }
+
+ public String nativeName() {
+ return nativeName;
+ }
+ }
+
protected ReadFileFormat fileFormat = ReadFileFormat.UnknownFormat;
private Boolean iterAsInput = false;
private StructType fileSchema;
private Map<String, String> fileReadProperties;
+ private ColumnMappingMode columnMappingMode;
LocalFilesNode(
Integer index,
@@ -114,6 +135,7 @@ public class LocalFilesNode implements SplitInfo {
this.fileReadProperties = other.fileReadProperties;
this.iterAsInput = other.iterAsInput;
this.fileSchema = other.fileSchema;
+ this.columnMappingMode = other.columnMappingMode;
this.otherMetadataColumns.addAll(otherMetadataColumns);
}
@@ -130,6 +152,14 @@ public class LocalFilesNode implements SplitInfo {
this.fileSchema = schema;
}
+ public void clearFileSchema() {
+ this.fileSchema = null;
+ }
+
+ public void setColumnMappingMode(ColumnMappingMode columnMappingMode) {
+ this.columnMappingMode = columnMappingMode;
+ }
+
private NamedStruct buildNamedStruct() {
NamedStruct.Builder namedStructBuilder = NamedStruct.newBuilder();
@@ -236,6 +266,14 @@ public class LocalFilesNode implements SplitInfo {
});
}
}
+ if (columnMappingMode != null) {
+
ReadRel.LocalFiles.FileOrFiles.otherConstantMetadataColumnValues.Builder
builder =
+
ReadRel.LocalFiles.FileOrFiles.otherConstantMetadataColumnValues.newBuilder();
+ builder
+ .setKey(COLUMN_MAPPING_MODE_METADATA_KEY)
+
.setValue(SubstraitUtil.convertJavaObjectToAny(columnMappingMode.nativeName()));
+ fileBuilder.addOtherConstMetadataColumns(builder.build());
+ }
switch (fileFormat) {
case ParquetReadFormat:
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 21a60b57bf..39d5acfb33 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
@@ -451,6 +451,8 @@ object GlutenConfig extends ConfigRegistry {
val SPARK_S3_AWS_IMDS_ENABLED: String = HADOOP_PREFIX + S3_AWS_IMDS_ENABLED
val ORC_FORCE_POSITIONAL_EVOLUTION = "orc.force.positional.evolution"
val SPARK_ORC_FORCE_POSITIONAL_EVOLUTION = HADOOP_PREFIX +
ORC_FORCE_POSITIONAL_EVOLUTION
+ val VELOX_PARQUET_USE_COLUMN_NAMES =
+ "spark.gluten.sql.columnar.backend.velox.parquetUseColumnNames"
// ABFS config
val ABFS_PREFIX = "fs.azure."
@@ -593,32 +595,20 @@ object GlutenConfig extends ConfigRegistry {
val confPrefixSession = prefixSessionOf(backendName)
val confPrefix = prefixOf(backendName)
+ // Column mapping mode is passed to Velox through scan splits, not through
+ // native session configs.
+ val veloxSplitColumnMappingConfigs = Set(VELOX_PARQUET_USE_COLUMN_NAMES)
conf
.filter {
case (k, _) =>
- // Backend's dynamic session conf only.
- k.startsWith(confPrefix) && !SQLConf.isStaticConfigKey(k) ||
- // put in all gluten velox configs
- k.startsWith(confPrefixSession)
+ val isBackendDynamicConf = k.startsWith(confPrefix) &&
!SQLConf.isStaticConfigKey(k)
+ val isBackendSessionConf = k.startsWith(confPrefixSession)
+ val isVeloxSplitColumnMappingConf =
+ backendName == "velox" &&
veloxSplitColumnMappingConfigs.contains(k)
+ (isBackendDynamicConf || isBackendSessionConf) &&
!isVeloxSplitColumnMappingConf
}
.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). 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.orcForcePositionalEvolution",
- "true")
- }
-
// Pass the latest tokens to native
nativeConfMap.put(
ReservedKeys.GLUTEN_UGI_TOKENS,
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 ddd61613ee..e17fcc70e2 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,48 @@ class GlutenHiveSQLQuerySuite extends
GlutenHiveSQLQuerySuiteBase {
}
}
+ testGluten("ORC positional and Parquet name mapping can coexist in one Velox
query") {
+ val hiveClient: HiveClient =
+
spark.sharedState.externalCatalog.unwrapped.asInstanceOf[HiveExternalCatalog].client
+
+ withSQLConf(
+ "spark.sql.hive.convertMetastoreOrc" -> "false",
+ "spark.sql.hive.convertMetastoreParquet" -> "false",
+ "spark.hadoop.orc.force.positional.evolution" -> "true"
+ ) {
+ withTempDir {
+ dir =>
+ val orcLoc = dir.toPath.resolve("test_orc_pos").toUri.toString
+ val parquetLoc =
dir.toPath.resolve("test_parquet_name").toUri.toString
+ withTable(
+ "test_orc_pos",
+ "test_orc_pos_renamed",
+ "test_parquet_name",
+ "test_parquet_name_reordered") {
+ hiveClient.runSqlHive(
+ s"create table test_orc_pos(c1 int, c2 int) stored as orc
location '$orcLoc'")
+ hiveClient.runSqlHive("insert into test_orc_pos select 1, 2")
+ hiveClient.runSqlHive(
+ s"create table test_orc_pos_renamed(x int, y int) stored as orc
location '$orcLoc'")
+
+ hiveClient.runSqlHive(
+ s"create table test_parquet_name(p int, q int) stored as parquet
" +
+ s"location '$parquetLoc'")
+ hiveClient.runSqlHive("insert into test_parquet_name select 3, 4")
+ hiveClient.runSqlHive(
+ s"create table test_parquet_name_reordered(q int, p int) stored
as parquet " +
+ s"location '$parquetLoc'")
+
+ val df = sql(
+ "select o.x, o.y, p.q, p.p from test_orc_pos_renamed o " +
+ "cross join test_parquet_name_reordered p")
+ checkAnswer(df, Seq(Row(1, 2, 4, 3)))
+ checkOperatorMatch[HiveTableScanExecTransformer](df)
+ }
+ }
+ }
+ }
+
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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
