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 9801f4169f [VL] Fix Parquet metadata scans with same-named data
columns (#13136)
9801f4169f is described below
commit 9801f4169f820b50d9a0e6d71b1f5e6e0085c320
Author: Felipe Pessoto <[email protected]>
AuthorDate: Thu Oct 29 13:32:03 2026 -0700
[VL] Fix Parquet metadata scans with same-named data columns (#13136)
---
.../util/delta-spark-ut/known-failures.txt | 12 -
cpp/velox/substrait/SubstraitToVeloxPlan.cc | 52 ++--
.../tests/Substrait2VeloxPlanConversionTest.cc | 243 +++++++++++++++++
.../GlutenFileMetadataStructSuite.scala | 293 ++++++++++++++++++++-
4 files changed, 557 insertions(+), 43 deletions(-)
diff --git a/.github/workflows/util/delta-spark-ut/known-failures.txt
b/.github/workflows/util/delta-spark-ut/known-failures.txt
index 6bb839c88f..9825b11cb3 100644
--- a/.github/workflows/util/delta-spark-ut/known-failures.txt
+++ b/.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
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2JsonSuite#data
skipping flags - old behavior with DataFrame schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2JsonSuite#data
skipping stats before and after optimize
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2JsonSuite#data
skipping stats before and after optimize - old behavior with DataFrame schema
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2ParquetSuite#Data
skipping handles aliasing for _metadata fields
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2ParquetSuite#Data
skipping handles aliasing for _metadata fields - old behavior with DataFrame
schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2ParquetSuite#data
skipping flags
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2ParquetSuite#data
skipping flags - old behavior with DataFrame schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1CheckpointV2ParquetSuite#data
skipping stats before and after optimize
@@ -385,22 +381,14 @@
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1ColMa
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1ColMapNameModeSuite#data
skipping flags - column mapping name mode - old behavior with DataFrame schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1ColMapNameModeSuite#data
skipping stats before and after optimize - column mapping name mode
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1ColMapNameModeSuite#data
skipping stats before and after optimize - column mapping name mode - old
behavior with DataFrame schema
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1Suite#Data
skipping handles aliasing for _metadata fields
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1Suite#Data
skipping handles aliasing for _metadata fields - old behavior with DataFrame
schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1Suite#data
skipping flags
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1Suite#data
skipping flags - old behavior with DataFrame schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1Suite#data
skipping stats before and after optimize
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1Suite#data
skipping stats before and after optimize - old behavior with DataFrame schema
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#Data
skipping handles aliasing for _metadata fields
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#Data
skipping handles aliasing for _metadata fields - old behavior with DataFrame
schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data
skipping flags
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch100Suite#data
skipping flags - old behavior with DataFrame schema
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#Data
skipping handles aliasing for _metadata fields
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#Data
skipping handles aliasing for _metadata fields - old behavior with DataFrame
schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data
skipping flags
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch1Suite#data
skipping flags - old behavior with DataFrame schema
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#Data
skipping handles aliasing for _metadata fields
-org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#Data
skipping handles aliasing for _metadata fields - old behavior with DataFrame
schema
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data
skipping flags
org.apache.spark.sql.delta.generatedsuites.dataskipping.DataSkippingDeltaV1WithCatalogOwnedBatch2Suite#data
skipping flags - old behavior with DataFrame schema
org.apache.spark.sql.delta.generatedsuites.delete.DeleteTempViewSQLNameBasedSuite#test
delete on temp view - nontrivial projection - Dataset TempView
diff --git a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
index 52e62bfadd..bb57cc6696 100644
--- a/cpp/velox/substrait/SubstraitToVeloxPlan.cc
+++ b/cpp/velox/substrait/SubstraitToVeloxPlan.cc
@@ -1656,34 +1656,48 @@ 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) {
+ // Scan outputs include connector-generated columns, not just physical
fields.
+ // For example, _metadata.file_name is VARCHAR even if the file contains a
+ // file_name BIGINT column. Exclude metadata here to avoid Parquet
type-checking
+ // that unrelated field before the connector supplies the metadata
constant.
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));
+ }
}
connector::ConnectorTableHandlePtr tableHandle;
diff --git a/cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc
b/cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc
index 323b6873fe..70a4894eac 100644
--- a/cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc
+++ b/cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc
@@ -17,11 +17,18 @@
#include "JsonToProtoConverter.h"
+#include <google/protobuf/wrappers.pb.h>
#include <filesystem>
+#include <limits>
#include "compute/VeloxPlanConverter.h"
+#include "operators/functions/RegistrationAllFunctions.h"
#include "substrait/SubstraitToVeloxPlan.h"
+#include "substrait/VeloxToSubstraitType.h"
#include "velox/common/base/tests/GTestUtils.h"
+#include "velox/dwio/common/FileSink.h"
#include "velox/dwio/common/tests/utils/DataFiles.h"
+#include "velox/dwio/parquet/RegisterParquetReader.h"
+#include "velox/dwio/parquet/writer/Writer.h"
#include "velox/exec/tests/utils/AssertQueryBuilder.h"
#include "velox/exec/tests/utils/HiveConnectorTestBase.h"
#include "velox/exec/tests/utils/TempDirectoryPath.h"
@@ -43,6 +50,17 @@ namespace gluten {
class Substrait2VeloxPlanConversionTest : public
exec::test::HiveConnectorTestBase {
protected:
+ void SetUp() override {
+ HiveConnectorTestBase::SetUp();
+ registerAllFunctions();
+ parquet::registerParquetReaderFactory();
+ }
+
+ void TearDown() override {
+ HiveConnectorTestBase::TearDown();
+ parquet::unregisterParquetReaderFactory();
+ }
+
#ifdef ENABLE_KAFKA
std::shared_ptr<SplitInfo> parseSerializedSplitInfo(const std::string&
serialized) {
return VeloxPlanConverter::parseSplitInfo(
@@ -75,12 +93,90 @@ 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);
+ // Velox's default whole-file range; the start offset defaults to zero.
+ file->set_length(std::numeric_limits<uint64_t>::max());
+ file->mutable_parquet();
+ auto* metadata = file->add_metadata_columns();
+ metadata->set_key("file_name");
+ metadata->set_value(std::filesystem::path(path).filename().string());
+ google::protobuf::StringValue mode;
+
mode.set_value(std::string(dwio::common::ColumnMappingModeName::toName(columnMappingMode)));
+ auto* mapping = file->add_other_const_metadata_columns();
+ mapping->set_key("__gluten.column_mapping_mode");
+ mapping->mutable_value()->PackFrom(mode);
+ if (tableSchema) {
+
file->mutable_schema()->CopyFrom(typeConverter.toSubstraitNamedStruct(arena,
tableSchema));
+ }
+ }
+ return files;
+ }
+
+ std::shared_ptr<const core::TableScanNode> convertRead(
+ const ::substrait::ReadRel& read,
+ const ::substrait::ReadRel_LocalFiles& files,
+ const std::vector<std::string>& functions = {}) {
+ ::substrait::Plan plan;
+ plan.add_relations()->mutable_rel()->mutable_read()->CopyFrom(read);
+ for (size_t i = 0; i < functions.size(); ++i) {
+ auto* function = plan.add_extensions()->mutable_extension_function();
+ function->set_function_anchor(static_cast<uint32_t>(i));
+ function->set_name(functions[i]);
+ }
+ planConverter_ = std::make_shared<VeloxPlanConverter>(
+ pool(),
+ veloxCfg_.get(),
+ std::vector<std::shared_ptr<ResultIterator>>{},
+ VeloxConnectorIds{.hive =
facebook::velox::exec::test::kHiveConnectorId});
+ auto scan = std::dynamic_pointer_cast<const
core::TableScanNode>(planConverter_->toVeloxPlan(plan, {files}));
+ VELOX_CHECK_NOT_NULL(scan);
+ return scan;
+ }
+
+ void writeParquet(const std::string& name, const RowVectorPtr& rows) {
+ const auto path = tmpDir_->getPath() + name;
+ auto file = std::make_unique<LocalWriteFile>(path, false, true);
+ auto sink = std::make_unique<dwio::common::WriteFileSink>(std::move(file),
path);
+ dwio::common::WriterOptions options;
+ options.memoryPool = rootPool_.get();
+ parquet::Writer writer(std::move(sink), options, asRowType(rows->type()));
+ writer.write(rows);
+ writer.close();
+ }
+
std::shared_ptr<exec::test::TempDirectoryPath>
tmpDir_{exec::test::TempDirectoryPath::create()};
std::shared_ptr<facebook::velox::config::ConfigBase> veloxCfg_ =
std::make_shared<facebook::velox::config::ConfigBase>(std::unordered_map<std::string,
std::string>());
@@ -301,6 +397,153 @@ TEST_F(Substrait2VeloxPlanConversionTest, filterUpper) {
planNode->toString(true, true));
}
+TEST_F(Substrait2VeloxPlanConversionTest, derivedFileSchemaUsesRegularSlots) {
+ auto read = makeRead(
+ ROW({"file_name", "p", "file_name", "row_index", "keep"}, {BIGINT(),
INTEGER(), VARCHAR(), BIGINT(), BOOLEAN()}),
+ {::substrait::NamedStruct::NORMAL_COL,
+ ::substrait::NamedStruct::PARTITION_COL,
+ ::substrait::NamedStruct::METADATA_COL,
+ ::substrait::NamedStruct::ROWINDEX_COL,
+ ::substrait::NamedStruct::NORMAL_COL});
+
read.mutable_filter()->mutable_selection()->mutable_direct_reference()->mutable_struct_field()->set_field(4);
+ auto scan = convertRead(read, makeParquetFiles({"/unused.parquet"}));
+ auto table = std::dynamic_pointer_cast<const
HiveTableHandle>(scan->tableHandle());
+ ASSERT_NE(table, nullptr);
+ EXPECT_EQ(table->dataColumns()->toString(),
"ROW<file_name:BIGINT,keep:BOOLEAN>");
+ EXPECT_EQ(scan->outputType()->size(), 5);
+ EXPECT_EQ(scan->assignments().size(), 5);
+ const std::vector<ColumnType> roles{
+ ColumnType::kRegular,
+ ColumnType::kPartitionKey,
+ ColumnType::kSynthesized,
+ ColumnType::kRowIndex,
+ ColumnType::kRegular};
+ for (int i = 0; i < roles.size(); ++i) {
+ auto column =
+ std::dynamic_pointer_cast<const
HiveColumnHandle>(scan->assignments().at(scan->outputType()->nameOf(i)));
+ ASSERT_NE(column, nullptr);
+ EXPECT_EQ(column->columnType(), roles[i]);
+ }
+ EXPECT_EQ(scan->outputType()->childAt(0)->kind(), TypeKind::BIGINT);
+ EXPECT_EQ(scan->outputType()->childAt(2)->kind(), TypeKind::VARCHAR);
+ auto filter = std::dynamic_pointer_cast<const
core::FieldAccessTypedExpr>(table->remainingFilter());
+ ASSERT_NE(filter, nullptr);
+ EXPECT_EQ(filter->name(), "keep");
+}
+
+TEST_F(Substrait2VeloxPlanConversionTest, emptyAndUntaggedFileSchemas) {
+ const auto type = ROW({"file_name", "row_index"}, {VARCHAR(), BIGINT()});
+ auto scan = convertRead(
+ makeRead(type, {::substrait::NamedStruct::METADATA_COL,
::substrait::NamedStruct::ROWINDEX_COL}),
+ makeParquetFiles({"/unused.parquet"}));
+ auto table = std::dynamic_pointer_cast<const
HiveTableHandle>(scan->tableHandle());
+ ASSERT_NE(table, nullptr);
+ ASSERT_NE(table->dataColumns(), nullptr);
+ EXPECT_EQ(table->dataColumns()->size(), 0);
+ EXPECT_EQ(scan->outputType()->size(), 2);
+ EXPECT_EQ(scan->assignments().size(), 2);
+
+ scan = convertRead(makeRead(type), makeParquetFiles({"/unused.parquet"}));
+ table = std::dynamic_pointer_cast<const
HiveTableHandle>(scan->tableHandle());
+ ASSERT_NE(table, nullptr);
+ EXPECT_TRUE(table->dataColumns()->equivalent(*type));
+}
+
+TEST_F(Substrait2VeloxPlanConversionTest,
explicitFileSchemaPreservesPhysicalFields) {
+ auto read = makeRead(
+ ROW({"FILE_NAME", "P", "ROW_INDEX", "VALUE"}, {VARCHAR(), INTEGER(),
BIGINT(), BIGINT()}),
+ {::substrait::NamedStruct::METADATA_COL,
+ ::substrait::NamedStruct::PARTITION_COL,
+ ::substrait::NamedStruct::ROWINDEX_COL,
+ ::substrait::NamedStruct::NORMAL_COL});
+ auto tableSchema = ROW({"FILE_NAME", "VALUE", "ROW_INDEX", "P"}, {BIGINT(),
BIGINT(), VARCHAR(), INTEGER()});
+ auto scan = convertRead(read, makeParquetFiles({"/unused.parquet"},
tableSchema));
+ auto table = std::dynamic_pointer_cast<const
HiveTableHandle>(scan->tableHandle());
+ ASSERT_NE(table, nullptr);
+ EXPECT_EQ(table->dataColumns()->toString(),
"ROW<file_name:BIGINT,value:BIGINT,row_index:VARCHAR>");
+ EXPECT_EQ(scan->outputType()->size(), 4);
+ EXPECT_EQ(scan->outputType()->childAt(0)->kind(), TypeKind::VARCHAR);
+ EXPECT_EQ(scan->outputType()->childAt(2)->kind(), TypeKind::BIGINT);
+}
+
+TEST_F(Substrait2VeloxPlanConversionTest, parquetMetadataOnlyPreservesRows) {
+ writeParquet("/first.parquet", makeRowVector({"file_name"},
{makeFlatVector<int64_t>({10, 20})}));
+ writeParquet("/second.parquet", makeRowVector({"file_name"},
{makeFlatVector<int64_t>({30})}));
+ const auto files = makeParquetFiles({"/first.parquet", "/second.parquet"});
+ auto scan = convertRead(makeRead(ROW({"file_name"}, {VARCHAR()}),
{::substrait::NamedStruct::METADATA_COL}), files);
+ EXPECT_EQ(planConverter_->splitInfos().at(scan->id())->columnMappingMode,
dwio::common::ColumnMappingMode::kName);
+ auto expected = makeRowVector({makeFlatVector<std::string>({"first.parquet",
"first.parquet", "second.parquet"})});
+
exec::test::AssertQueryBuilder(scan).splits(makeSplits(scan)).assertResults(expected);
+
+ // The same incompatible type is still an error when requested as physical
data.
+ scan = convertRead(makeRead(ROW({"file_name"}, {VARCHAR()})), files);
+ VELOX_ASSERT_THROW(
+
exec::test::AssertQueryBuilder(scan).splits(makeSplits(scan)).assertResults(expected),
+ "Converted type BIGINT is not allowed for requested type VARCHAR");
+}
+
+TEST_F(Substrait2VeloxPlanConversionTest, parquetMetadataAndRegularFilters) {
+ auto rows =
+ makeRowVector({"file_name", "keep"}, {makeFlatVector<int64_t>({10, 20}),
makeFlatVector<bool>({true, false})});
+ writeParquet("/first.parquet", rows);
+ writeParquet("/second.parquet", rows);
+ const auto files = makeParquetFiles({"/first.parquet", "/second.parquet"});
+ for (const std::string filename : {"first.parquet", "missing.parquet"}) {
+ auto read = makeRead(
+ ROW({"file_name", "keep"}, {VARCHAR(), BOOLEAN()}),
+ {::substrait::NamedStruct::METADATA_COL,
::substrait::NamedStruct::NORMAL_COL});
+ auto* conjunction = read.mutable_filter()->mutable_scalar_function();
+ conjunction->set_function_reference(1);
+ conjunction->mutable_output_type()->mutable_bool_()->set_nullability(
+ ::substrait::Type_Nullability_NULLABILITY_NULLABLE);
+ auto* equal =
conjunction->add_arguments()->mutable_value()->mutable_scalar_function();
+ equal->set_function_reference(0);
+ equal->mutable_output_type()->CopyFrom(conjunction->output_type());
+ equal->add_arguments()
+ ->mutable_value()
+ ->mutable_selection()
+ ->mutable_direct_reference()
+ ->mutable_struct_field()
+ ->set_field(0);
+
equal->add_arguments()->mutable_value()->mutable_literal()->set_string(filename);
+ conjunction->add_arguments()
+ ->mutable_value()
+ ->mutable_selection()
+ ->mutable_direct_reference()
+ ->mutable_struct_field()
+ ->set_field(1);
+ auto scan = convertRead(read, files, {"equal:opt_str_str",
"and:opt_bool_bool"});
+ EXPECT_EQ(planConverter_->splitInfos().at(scan->id())->columnMappingMode,
dwio::common::ColumnMappingMode::kName);
+ const int size = filename == "first.parquet" ? 1 : 0;
+ auto expected = makeRowVector({
+ makeFlatVector<std::string>(size, [&](auto /*row*/) { return filename;
}),
+ makeFlatVector<bool>(size, [](auto /*row*/) { return true; }),
+ });
+
exec::test::AssertQueryBuilder(scan).splits(makeSplits(scan)).assertResults(expected);
+ }
+}
+
+TEST_F(Substrait2VeloxPlanConversionTest,
parquetMetadataWithExplicitPositionalSchema) {
+ auto rows = makeRowVector(
+ {"file_name", "physical_keep"}, {makeFlatVector<int64_t>({10, 20}),
makeFlatVector<bool>({true, false})});
+ writeParquet("/first.parquet", rows);
+ const auto files = makeParquetFiles(
+ {"/first.parquet"},
+ ROW({"file_name", "keep"}, {BIGINT(), BOOLEAN()}),
+ dwio::common::ColumnMappingMode::kPosition);
+ auto scan = convertRead(
+ makeRead(
+ ROW({"file_name", "keep"}, {VARCHAR(), BOOLEAN()}),
+ {::substrait::NamedStruct::METADATA_COL,
::substrait::NamedStruct::NORMAL_COL}),
+ files);
+ EXPECT_EQ(planConverter_->splitInfos().at(scan->id())->columnMappingMode,
dwio::common::ColumnMappingMode::kPosition);
+ auto expected = makeRowVector({
+ makeFlatVector<std::string>({"first.parquet", "first.parquet"}),
+ makeFlatVector<bool>({true, false}),
+ });
+
exec::test::AssertQueryBuilder(scan).splits(makeSplits(scan)).assertResults(expected);
+}
+
TEST_F(Substrait2VeloxPlanConversionTest, expandSelectionMustBeTopLevelField) {
const auto makeExpandRel = [](bool nestedSelection) {
::substrait::Rel rel;
diff --git
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenFileMetadataStructSuite.scala
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenFileMetadataStructSuite.scala
index ed347d024c..16c129a695 100644
---
a/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenFileMetadataStructSuite.scala
+++
b/gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenFileMetadataStructSuite.scala
@@ -17,16 +17,26 @@
package org.apache.spark.sql.execution.datasources
import org.apache.gluten.backendsapi.BackendsApiManager
+import org.apache.gluten.config.GlutenConfig
import org.apache.gluten.execution.{FileSourceScanExecTransformer,
FilterExecTransformer}
import org.apache.spark.sql.{Column, DataFrame, Row}
import org.apache.spark.sql.GlutenSQLTestsBaseTrait
import org.apache.spark.sql.execution.FileSourceScanExec
-import org.apache.spark.sql.types.{IntegerType, LongType, StringType,
StructField, StructType}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DataType, IntegerType, LongType,
StringType, StructField, StructType}
+import org.apache.spark.util.Utils
+
+import org.apache.hadoop.fs.Path
+import org.apache.parquet.hadoop.ParquetFileReader
+import org.apache.parquet.hadoop.util.HadoopInputFile
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
import java.io.File
+import java.net.URI
import java.sql.Timestamp
+import scala.jdk.CollectionConverters._
import scala.reflect.ClassTag
class GlutenFileMetadataStructSuite extends FileMetadataStructSuite with
GlutenSQLTestsBaseTrait {
@@ -45,33 +55,76 @@ class GlutenFileMetadataStructSuite extends
FileMetadataStructSuite with GlutenS
private val METADATA_FILE_NAME = "_metadata.file_name"
private val METADATA_FILE_SIZE = "_metadata.file_size"
private val METADATA_FILE_MODIFICATION_TIME =
"_metadata.file_modification_time"
+ private val METADATA_FILE_BLOCK_START = "_metadata.file_block_start"
+ private val METADATA_FILE_BLOCK_LENGTH = "_metadata.file_block_length"
+
+ private val metadataColumns = Seq(
+ METADATA_FILE_PATH,
+ METADATA_FILE_NAME,
+ METADATA_FILE_SIZE,
+ METADATA_FILE_MODIFICATION_TIME,
+ METADATA_FILE_BLOCK_START,
+ METADATA_FILE_BLOCK_LENGTH)
+
+ private val schemaWithMetadataCollisions = {
+ val collisionTypes = Map[String, DataType](
+ METADATA_FILE_PATH -> LongType,
+ METADATA_FILE_NAME -> LongType,
+ METADATA_FILE_SIZE -> StringType,
+ METADATA_FILE_MODIFICATION_TIME -> StringType,
+ METADATA_FILE_BLOCK_START -> StringType,
+ METADATA_FILE_BLOCK_LENGTH -> StringType
+ )
+ StructType(
+ metadataColumns.map {
+ name => StructField(name.stripPrefix("_metadata."),
collisionTypes(name))
+ }).add("id", LongType)
+ }
+
+ private val idColumnIndex = schemaWithMetadataCollisions.fieldIndex("id")
+ private val maxSplitBytes = 1024L * 1024
+
+ private val collisionRows0 = Seq(
+ Row(10L, 20L, "size0", "time0", "start0", "length0", 0L),
+ Row(11L, 21L, "size1", "time1", "start1", "length1", 1L))
+ private val collisionRows1 = Seq(
+ Row(12L, 22L, "size2", "time2", "start2", "length2", 2L))
private def getMetadataForFile(f: File): Map[String, Any] = {
Map(
METADATA_FILE_PATH -> f.toURI.toString,
METADATA_FILE_NAME -> f.getName,
METADATA_FILE_SIZE -> f.length(),
- METADATA_FILE_MODIFICATION_TIME -> new Timestamp(f.lastModified())
+ METADATA_FILE_MODIFICATION_TIME -> new Timestamp(f.lastModified()),
+ METADATA_FILE_BLOCK_START -> 0L,
+ METADATA_FILE_BLOCK_LENGTH -> f.length()
)
}
- private def metadataColumnsNativeTest(testName: String, fileSchema:
StructType)(
+ private def metadataColumnsNativeTest(
+ testName: String,
+ fileSchema: StructType,
+ firstFileRows: Seq[Row] = data0,
+ secondFileRows: Seq[Row] = data1)(
f: (DataFrame, Map[String, Any], Map[String, Any]) => Unit): Unit = {
Seq("parquet").foreach {
testFileFormat =>
testGluten(s"metadata struct ($testFileFormat): " + testName) {
withTempDir {
dir =>
- import scala.collection.JavaConverters._
-
// 1. create df0 and df1 and save under /data/f0 and /data/f1
- val df0 = spark.createDataFrame(data0.asJava, fileSchema)
- val f0 = new File(dir, "data/f0").getCanonicalPath
- df0.coalesce(1).write.format(testFileFormat).save(f0)
+ // Spark's writer also produces valid Parquet footers for
zero-row fixtures.
+ withSQLConf(
+ GlutenConfig.GLUTEN_ENABLED.key -> "false",
+ GlutenConfig.NATIVE_WRITER_ENABLED.key -> "false") {
+ val df0 = spark.createDataFrame(firstFileRows.asJava,
fileSchema)
+ val f0 = new File(dir, "data/f0").getCanonicalPath
+ df0.coalesce(1).write.format(testFileFormat).save(f0)
- val df1 = spark.createDataFrame(data1.asJava, fileSchema)
- val f1 = new File(dir, "data/f1 gluten").getCanonicalPath
- df1.coalesce(1).write.format(testFileFormat).save(f1)
+ val df1 = spark.createDataFrame(secondFileRows.asJava,
fileSchema)
+ val f1 = new File(dir, "data/f1 gluten").getCanonicalPath
+ df1.coalesce(1).write.format(testFileFormat).save(f1)
+ }
// 2. read both f0 and f1
val df = spark.read
@@ -86,7 +139,10 @@ class GlutenFileMetadataStructSuite extends
FileMetadataStructSuite with GlutenS
.listFiles()
.filter(_.getName.endsWith(s".$testFileFormat"))
.head
- f(df, getMetadataForFile(realF0), getMetadataForFile(realF1))
+ assert(realF0.length() < maxSplitBytes && realF1.length() <
maxSplitBytes)
+ withSQLConf(SQLConf.FILES_MAX_PARTITION_BYTES.key ->
maxSplitBytes.toString) {
+ f(df, getMetadataForFile(realF0), getMetadataForFile(realF1))
+ }
}
}
}
@@ -97,6 +153,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(idColumnIndex)))
+ }
+ checkMetadataAnswer(selected, expected)
+ checkMetadataAnswer(
+ df.select(METADATA_FILE_NAME).distinct(),
+ Seq(Row(f0(METADATA_FILE_NAME)), Row(f1(METADATA_FILE_NAME))))
+ }
+
+ metadataColumnsNativeTest(
+ "whole metadata struct and struct alias with incompatible data fields",
+ schemaWithMetadataCollisions,
+ collisionRows0,
+ collisionRows1) {
+ (df, _, _) =>
+ val expected = withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false") {
+ df.select("_metadata").collect().toSeq
+ }
+ checkMetadataAnswer(df.select("_metadata"), expected, hasRowIndex = true)
+ checkMetadataAnswer(
+ df.select(Column("_metadata").as("aliased_metadata")),
+ expected,
+ hasRowIndex = true)
+ }
+
+ metadataColumnsNativeTest(
+ "metadata predicates do not use conflicting physical fields",
+ schemaWithMetadataCollisions,
+ collisionRows0,
+ collisionRows1) {
+ (df, f0, f1) =>
+ Seq(METADATA_FILE_NAME, METADATA_FILE_SIZE).foreach {
+ metadataColumn =>
+ val filtered = df
+ .where(Column(metadataColumn) === f0(metadataColumn) &&
Column("id") > 0L)
+ .select(METADATA_FILE_NAME, "id")
+ val expected = Seq((collisionRows0, f0), (collisionRows1, f1))
+ .filter { case (_, metadata) => metadata(metadataColumn) ==
f0(metadataColumn) }
+ .flatMap {
+ case (rows, metadata) =>
+ rows.filter(_.getLong(idColumnIndex) > 0L).map {
+ row => Row(metadata(METADATA_FILE_NAME),
row.getLong(idColumnIndex))
+ }
+ }
+ checkMetadataAnswer(filtered, expected)
+ }
+
+ // Direct timestamp synthesized-filter pushdown is not supported by the
native reader.
+ val targetSeconds = f0(METADATA_FILE_MODIFICATION_TIME)
+ .asInstanceOf[Timestamp]
+ .getTime / 1000L + 1L
+ val timestampFiltered = df
+ .where(
+ Column(METADATA_FILE_MODIFICATION_TIME).cast(LongType) +
Column("id") === targetSeconds)
+ .select(METADATA_FILE_NAME, "id")
+ val timestampExpected = Seq((collisionRows0, f0), (collisionRows1,
f1)).flatMap {
+ case (rows, metadata) =>
+ val seconds = metadata(METADATA_FILE_MODIFICATION_TIME)
+ .asInstanceOf[Timestamp]
+ .getTime / 1000L
+ rows.filter(row => seconds + row.getLong(idColumnIndex) ==
targetSeconds).map {
+ row => Row(metadata(METADATA_FILE_NAME),
row.getLong(idColumnIndex))
+ }
+ }
+ assert(timestampExpected.nonEmpty)
+ assert(timestampExpected.size < collisionRows0.size +
collisionRows1.size)
+ checkMetadataAnswer(timestampFiltered, timestampExpected)
+ if (BackendsApiManager.getSettings.supportNativeMetadataColumns()) {
+ checkOperatorMatch[FilterExecTransformer](timestampFiltered)
+ }
+ checkAnswer(
+ df.where(Column(METADATA_FILE_NAME) ===
"missing.parquet").select(METADATA_FILE_NAME),
+ Seq.empty)
+ }
+
+ metadataColumnsNativeTest(
+ "metadata collisions respect case normalization",
+ StructType(schemaWithMetadataCollisions.fields.map {
+ case field if field.name == "file_name" => field.copy(name = "FILE_NAME")
+ case field => field
+ }),
+ collisionRows0,
+ collisionRows1
+ ) {
+ (df, f0, f1) =>
+ Seq("false", "true").foreach {
+ caseSensitive =>
+ withSQLConf(SQLConf.CASE_SENSITIVE.key -> caseSensitive) {
+ checkMetadataAnswer(
+ df.select(METADATA_FILE_NAME),
+ Seq.fill(collisionRows0.size)(Row(f0(METADATA_FILE_NAME))) ++
+ Seq.fill(collisionRows1.size)(Row(f1(METADATA_FILE_NAME)))
+ )
+ }
+ }
+ }
+
+ metadataColumnsNativeTest(
+ "metadata with an explicit positional file schema",
+ schemaWithMetadataCollisions,
+ collisionRows0,
+ collisionRows1) {
+ (df, f0, f1) =>
+ withSQLConf(GlutenConfig.VELOX_PARQUET_USE_COLUMN_NAMES -> "false") {
+ checkMetadataAnswer(
+ df.select(METADATA_FILE_NAME),
+ Seq.fill(collisionRows0.size)(Row(f0(METADATA_FILE_NAME))) ++
+ Seq.fill(collisionRows1.size)(Row(f1(METADATA_FILE_NAME)))
+ )
+ checkMetadataAnswer(
+ df.select(METADATA_FILE_NAME, "id"),
+ collisionRows0.map(row => Row(f0(METADATA_FILE_NAME),
row.getLong(idColumnIndex))) ++
+ collisionRows1.map(row => Row(f1(METADATA_FILE_NAME),
row.getLong(idColumnIndex)))
+ )
+ }
+ }
+
+ testGluten("metadata constants with partition-only projection and pruning") {
+ withTempPath {
+ path =>
+ spark
+ .range(4)
+ .selectExpr("id AS file_name", "cast(id % 2 as int) AS p")
+ .coalesce(1)
+ .write
+ .partitionBy("p")
+ .parquet(path.getCanonicalPath)
+ val df = spark.read.parquet(path.getCanonicalPath)
+ val expected = withSQLConf(GlutenConfig.GLUTEN_ENABLED.key -> "false")
{
+ df.select(METADATA_FILE_NAME, "p").collect().toSeq
+ }
+ checkMetadataAnswer(df.select(METADATA_FILE_NAME), expected.map(row =>
Row(row.get(0))))
+ checkMetadataAnswer(df.select(METADATA_FILE_NAME, "p"), expected)
+ checkMetadataAnswer(
+ df.where(Column("p") === 1).select(METADATA_FILE_NAME, "p"),
+ expected.filter(_.getInt(1) == 1))
+ }
+ }
+
metadataColumnsNativeTest(
"plan check with metadata and user data select",
schemaWithFilePathField) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
