Copilot commented on code in PR #13136:
URL: https://github.com/apache/gluten/pull/13136#discussion_r4405418867
##########
.github/workflows/util/delta-spark-ut/known-failures.txt:
##########
@@ -369,14 +369,10 @@
org.apache.spark.sql.delta.deletionvectors.DeletionVectorsWithPredicatePushdownS
org.apache.spark.sql.delta.deletionvectors.DeletionVectorsWithPredicatePushdownSuite#PredicatePushdown:
Single deletion at the second row group. vectorizedReaderEnabled: true
readColumnarBatchAsRows: true
org.apache.spark.sql.delta.deletionvectors.DeletionVectorsWithPredicatePushdownSuite#huge
table: delete a small number of rows from tables of 2B rows with DVs
org.apache.spark.sql.delta.deletionvectors.DeletionVectorsWithPredicatePushdownSuite#huge
table: read from tables of 2B rows with existing DV of many zeros
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2JsonSuite#Data
skipping handles aliasing for _metadata fields
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2JsonSuite#Data
skipping handles aliasing for _metadata fields - old behavior with DataFrame
schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2JsonSuite#data
skipping flags
Review Comment:
The PR description says the 12 removed entries use renamed Delta 4.4.1 suite
identities, and that the fixes were validated in the Delta v4.2.0 / Spark 4.1
run, which predates the Delta upgrade. The diff only removes the existing
`DataSkippingDeltaV1*` identities and does not show any renamed identities.
Either the description or the removed identities are incorrect. Please confirm
which Delta version these identities target and re-run the Delta suite on that
version before removing the entries.
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -1656,34 +1656,44 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
auto names = colNameList;
auto types = veloxTypeList;
- // The columns we project from the file.
+ // The logical scan schema includes columns supplied by the connector, not
just the file.
auto baseSchema = ROW(std::move(names), std::move(types));
- // The columns present in the table, if not available default to the
baseSchema.
- auto tableSchema = splitInfo->tableSchema ? splitInfo->tableSchema :
baseSchema;
- // Build dataColumns from tableSchema, excluding partition columns.
// HiveTableHandle::dataColumns() is used as fileSchema for the reader.
- // Partition columns should not be validated against the file's physical
types
- // (their values come from the partition path, not from the file).
- std::unordered_set<std::string> partitionColNames;
- for (int idx = 0; idx < colNameList.size(); idx++) {
- if (columnTypes[idx] == ColumnType::kPartitionKey) {
- partitionColNames.insert(colNameList[idx]);
- }
- }
RowTypePtr dataColumns;
- if (partitionColNames.empty()) {
- dataColumns = tableSchema;
- } else {
+ if (!splitInfo->tableSchema) {
std::vector<std::string> dataColNames;
std::vector<TypePtr> dataColTypes;
- for (int idx = 0; idx < tableSchema->size(); idx++) {
- if (partitionColNames.find(tableSchema->nameOf(idx)) ==
partitionColNames.end()) {
- dataColNames.push_back(tableSchema->nameOf(idx));
- dataColTypes.push_back(tableSchema->childAt(idx));
+ for (int idx = 0; idx < colNameList.size(); idx++) {
+ if (columnTypes[idx] == ColumnType::kRegular) {
+ dataColNames.push_back(colNameList[idx]);
+ dataColTypes.push_back(veloxTypeList[idx]);
}
}
dataColumns = ROW(std::move(dataColNames), std::move(dataColTypes));
+ } else {
+ // An explicit schema describes physical fields and their ordinals. A field
+ // must not be removed just because a synthetic output has the same name.
+ const auto& tableSchema = splitInfo->tableSchema;
+ std::unordered_set<std::string> partitionColNames;
+ for (int idx = 0; idx < colNameList.size(); idx++) {
+ if (columnTypes[idx] == ColumnType::kPartitionKey) {
+ partitionColNames.insert(colNameList[idx]);
+ }
+ }
+ if (partitionColNames.empty()) {
+ dataColumns = tableSchema;
+ } else {
+ std::vector<std::string> dataColNames;
+ std::vector<TypePtr> dataColTypes;
+ for (int idx = 0; idx < tableSchema->size(); idx++) {
+ if (partitionColNames.find(tableSchema->nameOf(idx)) ==
partitionColNames.end()) {
+ dataColNames.push_back(tableSchema->nameOf(idx));
+ dataColTypes.push_back(tableSchema->childAt(idx));
+ }
+ }
+ dataColumns = ROW(std::move(dataColNames), std::move(dataColTypes));
+ }
Review Comment:
The comment on the explicit-schema branch says physical fields and their
ordinals must be preserved, but this loop drops any field whose name matches a
partition key. Removing a field from the middle of the schema shifts the
ordinals of later fields, which breaks positional mapping
(ColumnMappingMode::kPosition). Consider keeping every physical field in the
explicit schema and only excluding partition keys in a way that does not change
ordinals, such as marking them rather than removing them, or pass the
partition-exclusion decision to the reader so the physical schema stays intact.
##########
cpp/velox/substrait/SubstraitToVeloxPlan.cc:
##########
@@ -1656,34 +1656,44 @@ core::PlanNodePtr
SubstraitToVeloxPlanConverter::toVeloxPlan(const ::substrait::
auto names = colNameList;
auto types = veloxTypeList;
- // The columns we project from the file.
+ // The logical scan schema includes columns supplied by the connector, not
just the file.
auto baseSchema = ROW(std::move(names), std::move(types));
- // The columns present in the table, if not available default to the
baseSchema.
- auto tableSchema = splitInfo->tableSchema ? splitInfo->tableSchema :
baseSchema;
- // Build dataColumns from tableSchema, excluding partition columns.
// HiveTableHandle::dataColumns() is used as fileSchema for the reader.
- // Partition columns should not be validated against the file's physical
types
- // (their values come from the partition path, not from the file).
- std::unordered_set<std::string> partitionColNames;
- for (int idx = 0; idx < colNameList.size(); idx++) {
- if (columnTypes[idx] == ColumnType::kPartitionKey) {
- partitionColNames.insert(colNameList[idx]);
- }
- }
RowTypePtr dataColumns;
- if (partitionColNames.empty()) {
- dataColumns = tableSchema;
- } else {
+ if (!splitInfo->tableSchema) {
std::vector<std::string> dataColNames;
std::vector<TypePtr> dataColTypes;
- for (int idx = 0; idx < tableSchema->size(); idx++) {
- if (partitionColNames.find(tableSchema->nameOf(idx)) ==
partitionColNames.end()) {
- dataColNames.push_back(tableSchema->nameOf(idx));
- dataColTypes.push_back(tableSchema->childAt(idx));
+ for (int idx = 0; idx < colNameList.size(); idx++) {
+ if (columnTypes[idx] == ColumnType::kRegular) {
Review Comment:
The two branches each build dataColumns with a separate loop and similar
name and type collection logic. Consider a small helper, such as one that takes
a predicate over (name, type, index), so the filtering rules for regular and
partition columns live in one place and future column types only need to be
handled once.
##########
cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc:
##########
@@ -75,12 +93,89 @@ class Substrait2VeloxPlanConversionTest : public
exec::test::HiveConnectorTestBa
if (splitInfo->columnMappingMode.has_value()) {
splitBuilder.columnMappingMode(*splitInfo->columnMappingMode);
}
+ for (const auto& [name, value] : splitInfo->metadataColumns.at(i)) {
+ splitBuilder.infoColumn(name, value);
+ }
auto split = splitBuilder.build();
+ EXPECT_EQ(split->columnMappingMode, splitInfo->columnMappingMode);
splits.emplace_back(split);
}
return splits;
}
+ ::substrait::ReadRel makeRead(
+ const RowTypePtr& type,
+ const std::vector<::substrait::NamedStruct::ColumnType>& columnTypes =
{}) {
+ google::protobuf::Arena arena;
+ VeloxToSubstraitTypeConvertor typeConverter;
+ ::substrait::ReadRel read;
+ read.mutable_common()->mutable_direct();
+
read.mutable_base_schema()->CopyFrom(typeConverter.toSubstraitNamedStruct(arena,
type));
+ for (auto columnType : columnTypes) {
+ read.mutable_base_schema()->add_column_types(columnType);
+ }
+ return read;
+ }
+
+ ::substrait::ReadRel_LocalFiles makeParquetFiles(
+ const std::vector<std::string>& paths,
+ const RowTypePtr& tableSchema = nullptr,
+ dwio::common::ColumnMappingMode columnMappingMode =
dwio::common::ColumnMappingMode::kName) {
+ ::substrait::ReadRel_LocalFiles files;
+ google::protobuf::Arena arena;
+ VeloxToSubstraitTypeConvertor typeConverter;
+ for (const auto& path : paths) {
+ auto* file = files.add_items();
+ file->set_uri_file(path);
+ file->set_length(std::numeric_limits<uint64_t>::max());
Review Comment:
Setting the file length to the maximum `uint64_t` value is an unexplained
magic value. Add a comment describing what it means, such as 'length unknown,
let the reader determine it', or use a named constant so that future readers do
not assume it is a real file size.
##########
cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc:
##########
@@ -75,12 +93,89 @@ class Substrait2VeloxPlanConversionTest : public
exec::test::HiveConnectorTestBa
if (splitInfo->columnMappingMode.has_value()) {
splitBuilder.columnMappingMode(*splitInfo->columnMappingMode);
}
+ for (const auto& [name, value] : splitInfo->metadataColumns.at(i)) {
+ splitBuilder.infoColumn(name, value);
+ }
auto split = splitBuilder.build();
+ EXPECT_EQ(split->columnMappingMode, splitInfo->columnMappingMode);
splits.emplace_back(split);
}
return splits;
}
+ ::substrait::ReadRel makeRead(
+ const RowTypePtr& type,
+ const std::vector<::substrait::NamedStruct::ColumnType>& columnTypes =
{}) {
+ google::protobuf::Arena arena;
+ VeloxToSubstraitTypeConvertor typeConverter;
+ ::substrait::ReadRel read;
+ read.mutable_common()->mutable_direct();
+
read.mutable_base_schema()->CopyFrom(typeConverter.toSubstraitNamedStruct(arena,
type));
+ for (auto columnType : columnTypes) {
Review Comment:
The loop variable copies each element of the `columnTypes` vector. Iterate
by const reference (`for (const auto& columnType : columnTypes)`) to avoid
unnecessary copies.
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenFileMetadataStructSuite.scala:
##########
@@ -97,6 +150,219 @@ class GlutenFileMetadataStructSuite extends
FileMetadataStructSuite with GlutenS
assert(executedPlan.exists(plan => plan.getClass == tag.runtimeClass))
}
+ private def checkMetadataAnswer(
+ df: DataFrame,
+ expected: Seq[Row],
+ hasRowIndex: Boolean = false): Unit = {
+ checkAnswer(df, expected)
+ val settings = BackendsApiManager.getSettings
+ if (
+ settings.supportNativeMetadataColumns() &&
+ (!hasRowIndex || settings.supportNativeRowIndexColumn())
+ ) {
+ checkOperatorMatch[FileSourceScanExecTransformer](df)
+ } else {
+ checkOperatorMatch[FileSourceScanExec](df)
+ }
+ }
+
+ metadataColumns.zipWithIndex.foreach {
+ case (metadataColumn, index) =>
+ metadataColumnsNativeTest(
+ s"incompatible same-name data field for $metadataColumn",
+ schemaWithMetadataCollisions,
+ collisionRows0,
+ collisionRows1) {
+ (df, f0, f1) =>
+ val expected =
Seq.fill(collisionRows0.size)(Row(f0(metadataColumn))) ++
+ Seq.fill(collisionRows1.size)(Row(f1(metadataColumn)))
+ checkMetadataAnswer(df.select(metadataColumn), expected)
+
checkMetadataAnswer(df.select(Column(metadataColumn).as("aliased_metadata")),
expected)
+
+ val combined = df.select(
+ Column(metadataColumn.stripPrefix("_metadata.")).as("user_value"),
+ Column(metadataColumn).as("metadata_value"))
+ checkAnswer(
+ combined,
+ collisionRows0.map(row => Row(row.get(index), f0(metadataColumn)))
++
+ collisionRows1.map(row => Row(row.get(index),
f1(metadataColumn))))
+ checkOperatorMatch[FileSourceScanExec](combined)
+ }
+ }
+
+ metadataColumnsNativeTest(
+ "metadata-only projection of empty Parquet files",
+ schemaWithMetadataCollisions,
+ Seq.empty,
+ Seq.empty) {
+ (df, _, _) =>
+ assert(df.inputFiles.length == 2)
+ df.inputFiles.foreach {
+ file =>
+ val input =
+ HadoopInputFile.fromPath(new Path(new URI(file)),
spark.sessionState.newHadoopConf())
+ Utils.tryWithResource(ParquetFileReader.open(input)) {
+ reader =>
+ val footer = reader.getFooter
+ assert(footer.getBlocks.asScala.map(_.getRowCount).sum == 0L)
+ assert(
+ footer.getFileMetaData.getSchema
+ .asGroupType()
+ .getType("file_name")
+ .asPrimitiveType()
+ .getPrimitiveTypeName == PrimitiveTypeName.INT64)
+ }
+ }
+ checkMetadataAnswer(df.select(METADATA_FILE_NAME), Seq.empty)
+ }
+
+ metadataColumnsNativeTest(
+ "metadata constants with a regular sibling",
+ schemaWithMetadataCollisions,
+ collisionRows0,
+ collisionRows1) {
+ (df, f0, f1) =>
+ val selected = df.select((metadataColumns.map(Column(_)) :+
Column("id")): _*)
+ val expected = Seq((collisionRows0, f0), (collisionRows1, f1)).flatMap {
+ case (rows, metadata) =>
+ rows.map(row => Row.fromSeq(metadataColumns.map(metadata) :+
row.getLong(6)))
Review Comment:
The index 6 is an unexplained magic number for the `id` column. It is
repeated in several tests (for example lines 267, 268, 287, 288, 338, 339), so
a change to the fixture layout would silently break them. Define a named
constant such as `IdColumnIndex` (or look up the value by position from
`schemaWithMetadataCollisions`) and use it in each place.
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenFileMetadataStructSuite.scala:
##########
@@ -86,7 +136,10 @@ class GlutenFileMetadataStructSuite extends
FileMetadataStructSuite with GlutenS
.listFiles()
.filter(_.getName.endsWith(s".$testFileFormat"))
.head
- f(df, getMetadataForFile(realF0), getMetadataForFile(realF1))
+ assert(realF0.length() < 1024 * 1024 && realF1.length() < 1024 *
1024)
+ withSQLConf(SQLConf.FILES_MAX_PARTITION_BYTES.key -> "1m") {
Review Comment:
The 1 MB limit is repeated as a literal in two forms (`1024 * 1024` and
`\"1m\"`). They must stay in sync for the test to be meaningful, but nothing
links them. Extract a single constant, for example `MaxSplitBytes`, and derive
the config string and the assertion from it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
