felipepessoto commented on code in PR #13136:
URL: https://github.com/apache/gluten/pull/13136#discussion_r4353510761
##########
cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc:
##########
@@ -64,12 +82,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());
+ 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();
+ }
Review Comment:
The read path already adds the same prefix as the writer.
[`makeSplits()`](https://github.com/apache/gluten/blob/824caedfbd99fadf2e2c566c32d091b9f58e679b/cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc#L76-L81)
constructs the connector path with `fmt::format("{}{}", tmpDir_->getPath(),
paths[i])`, and every Parquet execution test uses `.splits(makeSplits(scan))`.
Both sides therefore use `<tmpDir>/first.parquet`, not `/first.parquet`.
Keeping the existing behavior; this repeats the path concern addressed in the
earlier resolved threads.
##########
cpp/velox/tests/Substrait2VeloxPlanConversionTest.cc:
##########
@@ -290,6 +385,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);
Review Comment:
This test passes `makeSplits(scan)` to `AssertQueryBuilder`. That helper
prepends `tmpDir_->getPath()` to the serialized path before creating the
connector split, matching `writeParquet()`. The execution does not open
`/first.parquet` at the filesystem root. Keeping the current code and resolving
this duplicate of the [path
discussion](https://github.com/apache/gluten/pull/13136#discussion_r4353510761).
--
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]
