feat: Add tableParameters to HiveInsertTableHandle (facebookincubator#16637)

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

Author: hdikeman

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>& tableParameters) {
   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,
+          .tableParameters = tableParameters,
       });
 }
 
@@ -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>& tableParameters)
     : inputColumns_(std::move(inputColumns)),
       locationHandle_(std::move(locationHandle)),
       storageFormat_(storageFormat),
@@ -440,6 +443,7 @@ HiveInsertTableHandle::HiveInsertTableHandle(
       writerOptions_(writerOptions),
       ensureFiles_(ensureFiles),
       fileNameGenerator_(std::move(fileNameGenerator)),
+      tableParameters_(tableParameters),
       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_->tableParameters()),
       options);
   return maybeCreateBucketSortWriter(writerIndex, std::move(writer));
 }
@@ -1369,6 +1374,13 @@ folly::dynamic HiveInsertTableHandle::serialize() const {
     params[key] = value;
   }
   obj["serdeParameters"] = params;
+
+  folly::dynamic tableParams = folly::dynamic::object;
+  for (const auto& [key, value] : tableParameters_) {
+    tableParams[key] = value;
+  }
+  obj["tableParameters"] = tableParams;
+
   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> tableParameters;
+  if (obj.count("tableParameters") > 0) {
+    for (const auto& pair : obj["tableParameters"].items()) {
+      tableParameters.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,
+      tableParameters);
 }
 
 void HiveInsertTableHandle::registerSerDe() {

velox/connectors/hive/HiveDataSink.h

@@ -254,7 +254,8 @@ 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>& tableParameters = {});
 
   virtual ~HiveInsertTableHandle() = default;
 
@@ -279,6 +280,10 @@ class HiveInsertTableHandle : public ConnectorInsertTableHandle {
     return serdeParameters_;
   }
 
+  const std::unordered_map<std::string, std::string>& tableParameters() const {
+    return tableParameters_;
+  }
+
   const std::shared_ptr<dwio::common::WriterOptions>& writerOptions() const {
     return writerOptions_;
   }
@@ -333,6 +338,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> tableParameters_;
   const std::vector<column_index_t> partitionChannels_;
   const std::vector<column_index_t> nonPartitionChannels_;
 };

velox/connectors/hive/tests/HiveConnectorSerDeTest.cpp

@@ -230,6 +230,11 @@ TEST_F(HiveConnectorSerDeTest, hiveInsertTableHandle) {
       {"key2", "value2"},
   };
 
+  std::unordered_map<std::string, std::string> tableParameters = {
+      {"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
+          tableParameters);
   testSerde(*hiveInsertTableHandle);
 }
 

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");

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> tableParameters{};
   };
 
   FileSink(std::string name, const Options& options)

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>& tableParameters) {
   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>(),
+      tableParameters);
 }
 
 std::shared_ptr<connector::hive::HiveColumnHandle>

velox/exec/tests/utils/HiveConnectorTestBase.h

@@ -207,7 +207,8 @@ 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>& tableParameters = {});
 
   static std::shared_ptr<connector::hive::HiveInsertTableHandle>
   makeHiveInsertTableHandle(

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->tableParameters());
 }
 
 std::shared_ptr<core::InsertTableHandle> createInsertTableHanlde(