feat: Add tableParameters to HiveInsertTableHandle (facebookincubator#16637)

hdikeman · meta-codesync[bot] · commit 8da2e573c247 · 2026-03-05T16:05:57.000-08:00
Summary: Pull Request resolved: facebookincubator#16637 Add a `tableParameters` field to `HiveInsertTableHandle` to carry table-level metadata from the coordinator to the native worker write path, following the existing `serdeParameters` pattern. - Add `tableParameters` constructor parameter, private member, and accessor to `HiveInsertTableHandle` serde changes. - Add `tableParameters` to `FileSink::Options`. - Thread through `createHiveFileSink` and both call sites. - Update all downstream callers and add serde test coverage. Differential Revision: D94986580
diff --git a/velox/connectors/hive/HiveDataSink.cpp b/velox/connectors/hive/HiveDataSink.cpp
@@ -70,7 +70,8 @@ std::unique_ptr<dwio::common::FileSink> createHiveFileSink(
     const std::shared_ptr<const HiveConfig>& hiveConfig,
     memory::MemoryPool* sinkPool,
     io::IoStatistics* ioStats,
-    IoStats* fileSystemStats) {
+    IoStats* fileSystemStats,
+    const std::unordered_map<std::string, std::string>& storageParameters) {
   return dwio::common::FileSink::create(
       path,
       {
@@ -81,6 +82,7 @@ std::unique_ptr<dwio::common::FileSink> createHiveFileSink(
           .metricLogger = dwio::common::MetricsLog::voidLog(),
           .stats = ioStats,
           .fileSystemStats = fileSystemStats,
+          .storageParameters = storageParameters,
       });
 }
 
@@ -430,7 +432,8 @@ HiveInsertTableHandle::HiveInsertTableHandle(
     // if there's no data. This is useful when the table is bucketed, but the
     // engine handles ensuring a 1 to 1 mapping from task to bucket.
     const bool ensureFiles,
-    std::shared_ptr<const FileNameGenerator> fileNameGenerator)
+    std::shared_ptr<const FileNameGenerator> fileNameGenerator,
+    const std::unordered_map<std::string, std::string>& storageParameters)
     : inputColumns_(std::move(inputColumns)),
       locationHandle_(std::move(locationHandle)),
       storageFormat_(storageFormat),
@@ -440,6 +443,7 @@ HiveInsertTableHandle::HiveInsertTableHandle(
       writerOptions_(writerOptions),
       ensureFiles_(ensureFiles),
       fileNameGenerator_(std::move(fileNameGenerator)),
+      storageParameters_(storageParameters),
       partitionChannels_(computePartitionChannels(inputColumns_)),
       nonPartitionChannels_(computeNonPartitionChannels(inputColumns_)) {
   if (compressionKind.has_value()) {
@@ -1077,7 +1081,8 @@ std::unique_ptr<dwio::common::Writer> HiveDataSink::createWriterForIndex(
           hiveConfig_,
           info->sinkPool.get(),
           ioStats_[writerIndex].get(),
-          fileSystemStats_.get()),
+          fileSystemStats_.get(),
+          insertTableHandle_->storageParameters()),
       options);
   return maybeCreateBucketSortWriter(writerIndex, std::move(writer));
 }
@@ -1369,6 +1374,13 @@ folly::dynamic HiveInsertTableHandle::serialize() const {
     params[key] = value;
   }
   obj["serdeParameters"] = params;
+
+  folly::dynamic storageParams = folly::dynamic::object;
+  for (const auto& [key, value] : storageParameters_) {
+    storageParams[key] = value;
+  }
+  obj["storageParameters"] = storageParams;
+
   obj["ensureFiles"] = ensureFiles_;
   obj["fileNameGenerator"] = fileNameGenerator_->serialize();
   return obj;
@@ -1400,6 +1412,13 @@ HiveInsertTableHandlePtr HiveInsertTableHandle::create(
     serdeParameters.emplace(pair.first.asString(), pair.second.asString());
   }
 
+  std::unordered_map<std::string, std::string> storageParameters;
+  if (obj.count("storageParameters") > 0) {
+    for (const auto& pair : obj["storageParameters"].items()) {
+      storageParameters.emplace(pair.first.asString(), pair.second.asString());
+    }
+  }
+
   bool ensureFiles = obj["ensureFiles"].asBool();
 
   auto fileNameGenerator =
@@ -1413,7 +1432,8 @@ HiveInsertTableHandlePtr HiveInsertTableHandle::create(
       serdeParameters,
       nullptr, // writerOptions is not serializable
       ensureFiles,
-      fileNameGenerator);
+      fileNameGenerator,
+      storageParameters);
 }
 
 void HiveInsertTableHandle::registerSerDe() {
diff --git a/velox/connectors/hive/HiveDataSink.h b/velox/connectors/hive/HiveDataSink.h
@@ -254,7 +254,9 @@ class HiveInsertTableHandle : public ConnectorInsertTableHandle {
       // engine handles ensuring a 1 to 1 mapping from task to bucket.
       const bool ensureFiles = false,
       std::shared_ptr<const FileNameGenerator> fileNameGenerator =
-          std::make_shared<const HiveInsertFileNameGenerator>());
+          std::make_shared<const HiveInsertFileNameGenerator>(),
+      const std::unordered_map<std::string, std::string>& storageParameters =
+          {});
 
   virtual ~HiveInsertTableHandle() = default;
 
@@ -275,10 +277,19 @@ class HiveInsertTableHandle : public ConnectorInsertTableHandle {
     return storageFormat_;
   }
 
+  /// Format specific options.
   const std::unordered_map<std::string, std::string>& serdeParameters() const {
     return serdeParameters_;
   }
 
+  /// Storage specific options.
+  const std::unordered_map<std::string, std::string>& storageParameters()
+      const {
+    return storageParameters_;
+  }
+
+  /// Avoid this in future usages. Format specific change should go through
+  /// serdeParameters.
   const std::shared_ptr<dwio::common::WriterOptions>& writerOptions() const {
     return writerOptions_;
   }
@@ -333,6 +344,7 @@ class HiveInsertTableHandle : public ConnectorInsertTableHandle {
   const std::shared_ptr<dwio::common::WriterOptions> writerOptions_;
   const bool ensureFiles_;
   const std::shared_ptr<const FileNameGenerator> fileNameGenerator_;
+  const std::unordered_map<std::string, std::string> storageParameters_;
   const std::vector<column_index_t> partitionChannels_;
   const std::vector<column_index_t> nonPartitionChannels_;
 };
diff --git a/velox/connectors/hive/tests/HiveConnectorSerDeTest.cpp b/velox/connectors/hive/tests/HiveConnectorSerDeTest.cpp
@@ -230,6 +230,11 @@ TEST_F(HiveConnectorSerDeTest, hiveInsertTableHandle) {
       {"key2", "value2"},
   };
 
+  std::unordered_map<std::string, std::string> storageParameters = {
+      {"key3", "value3"},
+      {"key4", "value4"},
+  };
+
   auto hiveInsertTableHandle =
       exec::test::HiveConnectorTestBase::makeHiveInsertTableHandle(
           tableColumnNames,
@@ -239,7 +244,10 @@ TEST_F(HiveConnectorSerDeTest, hiveInsertTableHandle) {
           locationHandle,
           dwio::common::FileFormat::NIMBLE,
           common::CompressionKind::CompressionKind_SNAPPY,
-          serdeParameters);
+          serdeParameters,
+          nullptr, // writerOptions
+          false, // ensureFiles
+          storageParameters);
   testSerde(*hiveInsertTableHandle);
 }
 
diff --git a/velox/connectors/hive/tests/HiveDataSinkTest.cpp b/velox/connectors/hive/tests/HiveDataSinkTest.cpp
@@ -167,7 +167,7 @@ class HiveDataSinkTest : public exec::test::HiveConnectorTestBase {
             connector::hive::LocationHandle::TableType::kNew),
         fileFormat,
         CompressionKind::CompressionKind_ZSTD,
-        {},
+        {}, // serdeParameters
         writerOptions,
         ensureFiles);
   }
@@ -1290,7 +1290,7 @@ TEST_F(HiveDataSinkTest, ensureFilesUnsupported) {
           dwio::common::FileFormat::DWRF,
           CompressionKind::CompressionKind_ZSTD,
           {}, // serdeParameters
-          nullptr, // writeOptions
+          nullptr, // writerOptions
           true // ensureFiles
           ),
       "ensureFiles is not supported with partition keys in the data");
@@ -1313,7 +1313,7 @@ TEST_F(HiveDataSinkTest, ensureFilesUnsupported) {
           dwio::common::FileFormat::DWRF,
           CompressionKind::CompressionKind_ZSTD,
           {}, // serdeParameters
-          nullptr, // writeOptions
+          nullptr, // writerOptions
           true // ensureFiles
           ),
       "ensureFiles is not supported with bucketing");
diff --git a/velox/dwio/common/FileSink.h b/velox/dwio/common/FileSink.h
@@ -47,6 +47,7 @@ class FileSink : public Closeable {
     MetricsLogPtr metricLogger{MetricsLog::voidLog()};
     IoStatistics* stats{nullptr};
     velox::IoStats* fileSystemStats{nullptr};
+    std::unordered_map<std::string, std::string> storageParameters{};
   };
 
   FileSink(std::string name, const Options& options)
diff --git a/velox/exec/tests/utils/HiveConnectorTestBase.cpp b/velox/exec/tests/utils/HiveConnectorTestBase.cpp
@@ -329,7 +329,7 @@ HiveConnectorTestBase::makeHiveInsertTableHandle(
       std::move(locationHandle),
       tableStorageFormat,
       compressionKind,
-      {},
+      {}, // serdeParameters
       writerOptions,
       ensureFiles);
 }
@@ -346,7 +346,8 @@ HiveConnectorTestBase::makeHiveInsertTableHandle(
     const std::optional<common::CompressionKind> compressionKind,
     const std::unordered_map<std::string, std::string>& serdeParameters,
     const std::shared_ptr<dwio::common::WriterOptions>& writerOptions,
-    const bool ensureFiles) {
+    const bool ensureFiles,
+    const std::unordered_map<std::string, std::string>& storageParameters) {
   std::vector<std::shared_ptr<const connector::hive::HiveColumnHandle>>
       columnHandles;
   std::vector<std::string> bucketedBy;
@@ -404,7 +405,9 @@ HiveConnectorTestBase::makeHiveInsertTableHandle(
       compressionKind,
       serdeParameters,
       writerOptions,
-      ensureFiles);
+      ensureFiles,
+      std::make_shared<const connector::hive::HiveInsertFileNameGenerator>(),
+      storageParameters);
 }
 
 std::shared_ptr<connector::hive::HiveColumnHandle>
diff --git a/velox/exec/tests/utils/HiveConnectorTestBase.h b/velox/exec/tests/utils/HiveConnectorTestBase.h
@@ -135,7 +135,7 @@ class HiveConnectorTestBase : public OperatorTestBase {
       const std::string& tableName = "hive_table",
       const RowTypePtr& dataColumns = nullptr,
       const std::vector<std::string>& indexColumns = {},
-      const std::unordered_map<std::string, std::string>& tableParameters =
+      const std::unordered_map<std::string, std::string>& storageParameters =
           {}) {
     return std::make_shared<connector::hive::HiveTableHandle>(
         kHiveConnectorId,
@@ -144,7 +144,7 @@ class HiveConnectorTestBase : public OperatorTestBase {
         remainingFilter,
         dataColumns,
         indexColumns,
-        tableParameters);
+        storageParameters);
   }
 
   /// @param name Column name.
@@ -207,7 +207,9 @@ class HiveConnectorTestBase : public OperatorTestBase {
       const std::unordered_map<std::string, std::string>& serdeParameters = {},
       const std::shared_ptr<dwio::common::WriterOptions>& writerOptions =
           nullptr,
-      const bool ensureFiles = false);
+      const bool ensureFiles = false,
+      const std::unordered_map<std::string, std::string>& storageParameters =
+          {});
 
   static std::shared_ptr<connector::hive::HiveInsertTableHandle>
   makeHiveInsertTableHandle(
diff --git a/velox/tool/trace/TableWriterReplayer.cpp b/velox/tool/trace/TableWriterReplayer.cpp
@@ -53,8 +53,11 @@ makeHiveInsertTableHandle(
           : std::make_shared<connector::hive::HiveBucketProperty>(
                 *tracedHandle->bucketProperty()),
       compressionKind,
-      std::unordered_map<std::string, std::string>{},
-      writerOptions);
+      std::unordered_map<std::string, std::string>{}, // serdeParameters
+      writerOptions,
+      tracedHandle->ensureFiles(),
+      tracedHandle->fileNameGenerator(),
+      tracedHandle->storageParameters());
 }
 
 std::shared_ptr<core::InsertTableHandle> createInsertTableHanlde(
