Resolve comments

velox/connectors/hive/HiveDataSink.cpp

@@ -38,8 +38,8 @@ namespace facebook::velox::connector::hive {
 
 namespace {
 
-// Returns the type corresponding to non-partition keys.
-RowTypePtr getDataType(
+// Returns the type of non-partition keys.
+RowTypePtr getNonPartitionTypes(
     const std::vector<column_index_t>& dataCols,
     const RowTypePtr& inputType) {
   std::vector<std::string> childNames;
@@ -57,6 +57,7 @@ RowTypePtr getDataType(
   return ROW(std::move(childNames), std::move(childTypes));
 }
 
+// Filters out partition columns if there is any.
 RowVectorPtr makeDataInput(
     const std::vector<column_index_t>& dataCols,
     const RowVectorPtr& input) {
@@ -70,7 +71,7 @@ RowVectorPtr makeDataInput(
 
   return std::make_shared<RowVector>(
       input->pool(),
-      getDataType(dataCols, asRowType(input->type())),
+      getNonPartitionTypes(dataCols, asRowType(input->type())),
       input->nulls(),
       input->size(),
       std::move(childVectors),
@@ -92,8 +93,8 @@ std::vector<column_index_t> getPartitionChannels(
   return channels;
 }
 
-// Returns a subset of column indices corresponding to non-partition keys.
-std::vector<column_index_t> getDataChannels(
+// Returns the column indices of non-partition keys.
+std::vector<column_index_t> getNonPartitionChannels(
     const std::vector<column_index_t>& partitionChannels,
     const column_index_t childrenSize) {
   std::vector<column_index_t> dataChannels;
@@ -366,7 +367,7 @@ HiveDataSink::HiveDataSink(
                     hiveConfig_->isFileColumnNamesReadAsLowerCase(
                         connectorQueryCtx->sessionProperties()))
               : nullptr),
-      dataChannels_(getDataChannels(partitionChannels_, inputType_->size())),
+      dataChannels_(getNonPartitionChannels(partitionChannels_, inputType_->size())),
       bucketCount_(
           insertTableHandle_->bucketProperty() == nullptr
               ? 0
@@ -400,7 +401,7 @@ HiveDataSink::HiveDataSink(
     sortCompareFlags_.reserve(sortedProperty.size());
     for (int i = 0; i < sortedProperty.size(); ++i) {
       auto columnIndex =
-          getDataType(dataChannels_, inputType_)
+          getNonPartitionTypes(dataChannels_, inputType_)
               ->getChildIdxIfExists(sortedProperty.at(i)->sortColumn());
       if (columnIndex.has_value()) {
         sortColumnIndices_.push_back(columnIndex.value());
@@ -664,7 +665,7 @@ uint32_t HiveDataSink::appendWriter(const HiveWriterId& id) {
   dwio::common::WriterOptions options;
   const auto* connectorSessionProperties =
       connectorQueryCtx_->sessionProperties();
-  options.schema = getDataType(dataChannels_, inputType_);
+  options.schema = getNonPartitionTypes(dataChannels_, inputType_);
 
   options.memoryPool = writerInfo_.back()->writerPool.get();
   options.compressionKind = insertTableHandle_->compressionKind();
@@ -711,7 +712,7 @@ HiveDataSink::maybeCreateBucketSortWriter(
   auto* sortPool = writerInfo_.back()->sortPool.get();
   VELOX_CHECK_NOT_NULL(sortPool);
   auto sortBuffer = std::make_unique<exec::SortBuffer>(
-      getDataType(dataChannels_, inputType_),
+      getNonPartitionTypes(dataChannels_, inputType_),
       sortColumnIndices_,
       sortCompareFlags_,
       sortPool,

velox/exec/tests/TableWriteTest.cpp

@@ -108,7 +108,7 @@ std::function<PlanNodePtr(std::string, PlanNodePtr)> addTableWriter(
   };
 }
 
-RowTypePtr getScanOutput(
+RowTypePtr getNonPartitionsColumns(
     const std::vector<std::string>& partitionedKeys,
     const RowTypePtr& rowType) {
   std::vector<std::string> dataColumnNames;
@@ -1511,7 +1511,7 @@ TEST_P(AllTableWriterTest, scanFilterProjectWrite) {
   // We create a new plan that only read that file and then
   // compare that against a duckDB query that runs the whole query.
   if (partitionedBy_.size() > 0) {
-    auto newOutputType = getScanOutput(partitionedBy_, outputType);
+    auto newOutputType = getNonPartitionsColumns(partitionedBy_, outputType);
     assertQuery(
         PlanBuilder().tableScan(newOutputType).planNode(),
         makeHiveConnectorSplits(outputDirectory),
@@ -1573,7 +1573,7 @@ TEST_P(AllTableWriterTest, renameAndReorderColumns) {
   assertQueryWithWriterConfigs(plan, filePaths, "SELECT count(*) FROM tmp");
 
   if (partitionedBy_.size() > 0) {
-    auto newOutputType = getScanOutput(partitionedBy_, tableSchema_);
+    auto newOutputType = getNonPartitionsColumns(partitionedBy_, tableSchema_);
     HiveConnectorTestBase::assertQuery(
         PlanBuilder().tableScan(newOutputType).planNode(),
         makeHiveConnectorSplits(outputDirectory),
@@ -1619,7 +1619,7 @@ TEST_P(AllTableWriterTest, directReadWrite) {
   // compare that against a duckDB query that runs the whole query.
 
   if (partitionedBy_.size() > 0) {
-    auto newOutputType = getScanOutput(partitionedBy_, tableSchema_);
+    auto newOutputType = getNonPartitionsColumns(partitionedBy_, tableSchema_);
     assertQuery(
         PlanBuilder().tableScan(newOutputType).planNode(),
         makeHiveConnectorSplits(outputDirectory),
@@ -1660,7 +1660,7 @@ TEST_P(AllTableWriterTest, constantVectors) {
   assertQuery(op, fmt::format("SELECT {}", size));
 
   if (partitionedBy_.size() > 0) {
-    auto newOutputType = getScanOutput(partitionedBy_, tableSchema_);
+    auto newOutputType = getNonPartitionsColumns(partitionedBy_, tableSchema_);
     assertQuery(
         PlanBuilder().tableScan(newOutputType).planNode(),
         makeHiveConnectorSplits(outputDirectory),
@@ -1719,7 +1719,7 @@ TEST_P(AllTableWriterTest, commitStrategies) {
     assertQuery(plan, "SELECT count(*) FROM tmp");
 
     if (partitionedBy_.size() > 0) {
-      auto newOutputType = getScanOutput(partitionedBy_, tableSchema_);
+      auto newOutputType = getNonPartitionsColumns(partitionedBy_, tableSchema_);
       assertQuery(
           PlanBuilder().tableScan(newOutputType).planNode(),
           makeHiveConnectorSplits(outputDirectory),
@@ -1755,7 +1755,7 @@ TEST_P(AllTableWriterTest, commitStrategies) {
     assertQuery(plan, "SELECT count(*) FROM tmp");
 
     if (partitionedBy_.size() > 0) {
-      auto newOutputType = getScanOutput(partitionedBy_, tableSchema_);
+      auto newOutputType = getNonPartitionsColumns(partitionedBy_, tableSchema_);
       assertQuery(
           PlanBuilder().tableScan(newOutputType).planNode(),
           makeHiveConnectorSplits(outputDirectory),
@@ -1950,7 +1950,7 @@ TEST_P(PartitionedTableWriterTest, multiplePartitions) {
   // Verify distribution of records in partition directories.
   auto iterPartitionDirectory = actualPartitionDirectories.begin();
   auto iterPartitionName = partitionNames.begin();
-  auto newOutputType = getScanOutput(partitionKeys, rowType);
+  auto newOutputType = getNonPartitionsColumns(partitionKeys, rowType);
   while (iterPartitionDirectory != actualPartitionDirectories.end()) {
     assertQuery(
         PlanBuilder().tableScan(newOutputType).planNode(),
@@ -2025,7 +2025,7 @@ TEST_P(PartitionedTableWriterTest, singlePartition) {
       fs::path(outputDirectory->path) / "p0=365");
 
   // Verify all data is written to the single partition directory.
-  auto newOutputType = getScanOutput(partitionKeys, rowType);
+  auto newOutputType = getNonPartitionsColumns(partitionKeys, rowType);
   assertQuery(
       PlanBuilder().tableScan(newOutputType).planNode(),
       makeHiveConnectorSplits(outputDirectory),
@@ -2072,7 +2072,7 @@ TEST_P(PartitionedWithoutBucketTableWriterTest, fromSinglePartitionToMultiple) {
 
   assertQueryWithWriterConfigs(plan, "SELECT count(*) FROM tmp");
 
-  auto newOutputType = getScanOutput(partitionKeys, rowType);
+  auto newOutputType = getNonPartitionsColumns(partitionKeys, rowType);
   assertQuery(
       PlanBuilder().tableScan(newOutputType).planNode(),
       makeHiveConnectorSplits(outputDirectory),
@@ -2416,7 +2416,7 @@ TEST_P(BucketedTableOnlyWriteTest, bucketCountLimit) {
       assertQueryWithWriterConfigs(plan, "SELECT count(*) FROM tmp");
 
       if (partitionedBy_.size() > 0) {
-        auto newOutputType = getScanOutput(partitionedBy_, tableSchema_);
+        auto newOutputType = getNonPartitionsColumns(partitionedBy_, tableSchema_);
         assertQuery(
             PlanBuilder().tableScan(newOutputType).planNode(),
             makeHiveConnectorSplits(outputDirectory),
@@ -3206,7 +3206,7 @@ TEST_P(BucketSortOnlyTableWriterTest, sortWriterSpill) {
   auto task =
       assertQueryWithWriterConfigs(op, fmt::format("SELECT {}", 5 * 500), true);
   if (partitionedBy_.size() > 0) {
-    rowType_ = getScanOutput(partitionedBy_, rowType_);
+    rowType_ = getNonPartitionsColumns(partitionedBy_, rowType_);
     verifyTableWriterOutput(outputDirectory->path, rowType_);
   } else {
     verifyTableWriterOutput(outputDirectory->path, rowType_);