Copilot commented on code in PR #13136:
URL: https://github.com/apache/gluten/pull/13136#discussion_r4353469946
##########
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:
writeParquet()` writes to `tmpDir_->getPath() + name`, but
`makeParquetFiles()` is being given absolute-looking paths like
`"/first.parquet"`, which will typically not point to the temp directory file
just written (and may attempt to read from filesystem root). Make
`writeParquet()` return the full path and pass that into `makeParquetFiles()`,
or have `makeParquetFiles()` join `tmpDir_->getPath()` with relative file names
(and avoid leading '/' in `name` to prevent path concatenation surprises).
##########
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:
writeParquet()` writes to `tmpDir_->getPath() + name`, but
`makeParquetFiles()` is being given absolute-looking paths like
`"/first.parquet"`, which will typically not point to the temp directory file
just written (and may attempt to read from filesystem root). Make
`writeParquet()` return the full path and pass that into `makeParquetFiles()`,
or have `makeParquetFiles()` join `tmpDir_->getPath()` with relative file names
(and avoid leading '/' in `name` to prevent path concatenation surprises).
##########
gluten-ut/spark41/src/test/scala/org/apache/spark/sql/execution/datasources/GlutenFileMetadataStructSuite.scala:
##########
@@ -45,33 +55,64 @@ 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 = StructType(
+ metadataColumns.zipWithIndex.map {
+ case (name, index) =>
+ StructField(name.stripPrefix("_metadata."), if (index < 2) LongType
else StringType)
Review Comment:
schemaWithMetadataCollisions` relies on `zipWithIndex` ordering (`index <
2`) to decide types. This is brittle: if `metadataColumns` order changes (or
new columns are inserted earlier), the collision types will silently change
(e.g., `_metadata.file_path` currently becomes `LongType` only due to
ordering). Prefer an explicit per-column mapping (e.g., a `Map[String,
DataType]` keyed by metadata column name) so the intended collision type for
each field is stable and self-documenting.
--
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]
