Iceberg core code

Alchemy-item: (ID = 1153) Iceberg staging hub commit 1/6 - c5a69de3d1021073c13a99e1c7c6d6fcce355178

refactor: Move toValues from InPredicate.cpp to Filter.h

The function toValues removes duplicated values from the vector and
return them in a std::vector. It was used to build an InPredicate. It
will be needed for building NOT IN filters for Iceberg equality delete
read as well, therefore moving it from velox/functions/prestosql/InPred
icate.cpp to velox/type/Filter.h. This commit also renames it to
deDuplicateValues to make it easier to understand.

feat(connector): Support reading Iceberg split with equality deletes

This commit introduces EqualityDeleteFileReader, which is used to read
Iceberg splits with equality delete files. The equality delete files
are read to construct domain filters or filter functions, which then
would be evaluated in the base file readers.

When there is only one equality delete field, and when that field is
an Iceberg identifier field, i.e. non-floating point primitive types,
the values would be converted to a list as a NOT IN domain filter,
with the NULL treated separately. This domain filter would then be
pushed to the ColumnReaders to filter our unwanted rows before they
are read into Velox vectors. When the equality delete column is a
nested column, e.g. a sub-column in a struct, or the key in a map,
such column may not be in the base file ScanSpec. We need to add/remove
these subfields to/from the SchemaWithId and ScanSpec recursively if
they were not in the ScanSpec already. A test is also added for such
case.

If there are more than one equality delete field, or the field is not
an Iceberg identifier field, the values would be converted to a typed
expression in the conjunct of disconjunts form. This expression would
be evaluated as the remaining filter function after the rows are read
into the Velox vectors. Note that this only works for Presto now as
the "neq" function is not registered by Spark. See https://github.com/
facebookincubator/issues/12667

Note that this commit only supports integral types. VARCHAR and
VARBINARY need to be supported in future commits (see
facebookincubator#12664).

Co-authored-by: Naveen Kumar Mahadevuni <Naveen.Mahadevuni@ibm.com>

Alchemy-item: (ID = 1153) Iceberg staging hub commit 2/6 - 14edb98c67f1c572a5f40682923795bd5b08e7c3

Support insert data into iceberg table.

Add iceberg partition transforms.

Co-authored-by: Chengcheng Jin <Chengcheng.Jin@ibm.com>

Add NaN statistics to parquet writer.

Collect Iceberg data file statistics in dwio.

Integrate Iceberg data file statistics and adding unit test.

Support write field_id to parquet metadata SchemaElement.

Implement iceberg sort order

Add clustered Iceberg writer mode.

Fix parquet writer ut

Add IcebergConnector

Fix unittest error

Resolve confict

Resolve confict

Fix test build issue

Fix crash

Author: PingLiuPing

velox/connectors/hive/CMakeLists.txt

@@ -31,7 +31,7 @@ velox_add_library(
   HiveDataSource.cpp
   FileIndexReader.cpp
   HiveIndexSource.cpp
-  HivePartitionName.cpp
+  HivePartitionUtil.cpp
   PartitionIdGenerator.cpp
   SplitReader.cpp
   TableHandle.cpp
@@ -45,6 +45,7 @@ velox_link_libraries(
     velox_dwio_catalog_fbhive
     velox_hive_partition_function
     velox_key_encoder
+  PUBLIC velox_hive_iceberg_splitreader
 )
 
 velox_add_library(velox_hive_partition_function HivePartitionFunction.cpp)

velox/connectors/hive/HiveConfig.cpp

@@ -316,4 +316,9 @@ std::string HiveConfig::schema(const config::ConfigBase* session) const {
       kSchema, config_->get<std::string>(kSchema, ""));
 }
 
+bool HiveConfig::fanoutEnabled(const config::ConfigBase* session) const {
+  return session->get<bool>(
+      kFanoutEnabledSession, config_->get<bool>(kFanoutEnabled, true));
+}
+
 } // namespace facebook::velox::connector::hive

velox/connectors/hive/HiveConfig.h

@@ -254,6 +254,12 @@ class HiveConfig {
   static constexpr const char* kSource = "source";
   static constexpr const char* kSchema = "schema";
 
+  /// Controls the writer mode, whether the fanout mode writer is enabled,
+  /// default value is true, setting to false means clustered mode.
+  /// Currently applies only to the Iceberg writer.
+  static constexpr const char* kFanoutEnabled = "fanout-enabled";
+  static constexpr const char* kFanoutEnabledSession = "fanout_enabled";
+
   InsertExistingPartitionsBehavior insertExistingPartitionsBehavior(
       const config::ConfigBase* session) const;
 
@@ -364,6 +370,9 @@ class HiveConfig {
   /// Schema of the query. Used for storage logging.
   std::string schema(const config::ConfigBase* session) const;
 
+  /// Return if fanout writer mode is enabled.
+  bool fanoutEnabled(const config::ConfigBase* session) const;
+
   HiveConfig(std::shared_ptr<const config::ConfigBase> config) {
     VELOX_CHECK_NOT_NULL(
         config, "Config is null for HiveConfig initialization");

velox/connectors/hive/HiveConnector.cpp

@@ -57,7 +57,7 @@ HiveConnector::HiveConnector(
 std::unique_ptr<DataSource> HiveConnector::createDataSource(
     const RowTypePtr& outputType,
     const ConnectorTableHandlePtr& tableHandle,
-    const ColumnHandleMap& columnHandles,
+    const std::unordered_map<std::string, ColumnHandlePtr>& columnHandles,
     ConnectorQueryCtx* connectorQueryCtx) {
   return std::make_unique<HiveDataSource>(
       outputType,

velox/connectors/hive/HiveConnectorUtil.cpp

@@ -25,6 +25,10 @@
 #include "velox/expression/ExprToSubfieldFilter.h"
 #include "velox/expression/FieldReference.h"
 
+#include <boost/lexical_cast.hpp>
+#include <boost/uuid/uuid_generators.hpp>
+#include <boost/uuid/uuid_io.hpp>
+
 namespace facebook::velox::connector::hive {
 namespace {
 
@@ -884,7 +888,6 @@ std::unique_ptr<dwio::common::BufferedInput> createBufferedInput(
 }
 
 namespace {
-
 core::CallTypedExprPtr replaceInputs(
     const core::CallTypedExpr* call,
     std::vector<core::TypedExprPtr>&& inputs) {
@@ -1057,6 +1060,10 @@ core::TypedExprPtr extractFiltersFromRemainingFilter(
 }
 } // namespace
 
+std::string makeUuid() {
+  return boost::lexical_cast<std::string>(boost::uuids::random_generator()());
+}
+
 core::TypedExprPtr extractFiltersFromRemainingFilter(
     const core::TypedExprPtr& expr,
     core::ExpressionEvaluator* evaluator,

velox/connectors/hive/HiveConnectorUtil.h

@@ -256,4 +256,6 @@ std::unique_ptr<common::Filter> createRangeFilter(
     const variant& lower,
     const variant& upper);
 
+std::string makeUuid();
+
 } // namespace facebook::velox::connector::hive

velox/connectors/hive/HiveDataSink.cpp

@@ -120,23 +120,34 @@ RowVectorPtr makeDataInput(
       input->getNullCount());
 }
 
-// Creates a PartitionIdGenerator if the table is partitioned, otherwise returns
-// nullptr.
-std::unique_ptr<PartitionIdGenerator> createPartitionIdGenerator(
-    const RowTypePtr& inputType,
-    const std::shared_ptr<const HiveInsertTableHandle>& insertTableHandle,
-    const std::shared_ptr<const HiveConfig>& hiveConfig,
-    const ConnectorQueryCtx* connectorQueryCtx) {
-  auto partitionChannels = insertTableHandle->partitionChannels();
-  if (partitionChannels.empty()) {
-    return nullptr;
+// Returns a subset of column indices corresponding to partition keys.
+std::vector<column_index_t> getPartitionChannels(
+    const std::shared_ptr<const HiveInsertTableHandle>& insertTableHandle) {
+  std::vector<column_index_t> channels;
+
+  for (column_index_t i = 0; i < insertTableHandle->inputColumns().size();
+       i++) {
+    if (insertTableHandle->inputColumns()[i]->isPartitionKey()) {
+      channels.push_back(i);
+    }
   }
-  return std::make_unique<PartitionIdGenerator>(
-      inputType,
-      partitionChannels,
-      hiveConfig->maxPartitionsPerWriters(
-          connectorQueryCtx->sessionProperties()),
-      connectorQueryCtx->memoryPool());
+
+  return channels;
+}
+
+// Returns the column indices of non-partition data columns.
+std::vector<column_index_t> getNonPartitionChannels(
+    const std::shared_ptr<const HiveInsertTableHandle>& insertTableHandle) {
+  std::vector<column_index_t> dataChannels;
+
+  for (column_index_t i = 0; i < insertTableHandle->inputColumns().size();
+       i++) {
+    if (!insertTableHandle->inputColumns()[i]->isPartitionKey()) {
+      dataChannels.push_back(i);
+    }
+  }
+
+  return dataChannels;
 }
 
 std::string makePartitionDirectory(
@@ -148,10 +159,6 @@ std::string makePartitionDirectory(
   return tableDirectory;
 }
 
-std::string makeUuid() {
-  return boost::lexical_cast<std::string>(boost::uuids::random_generator()());
-}
-
 std::unordered_map<LocationHandle::TableType, std::string> tableTypeNames() {
   return {
       {LocationHandle::TableType::kNew, "kNew"},
@@ -418,52 +425,6 @@ std::string HiveBucketProperty::toString() const {
   return out.str();
 }
 
-HiveInsertTableHandle::HiveInsertTableHandle(
-    std::vector<std::shared_ptr<const HiveColumnHandle>> inputColumns,
-    std::shared_ptr<const LocationHandle> locationHandle,
-    dwio::common::FileFormat storageFormat,
-    std::shared_ptr<const HiveBucketProperty> bucketProperty,
-    std::optional<common::CompressionKind> compressionKind,
-    const std::unordered_map<std::string, std::string>& serdeParameters,
-    const std::shared_ptr<dwio::common::WriterOptions>& writerOptions,
-    // When this option is set the HiveDataSink will always write a file even
-    // 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)
-    : inputColumns_(std::move(inputColumns)),
-      locationHandle_(std::move(locationHandle)),
-      storageFormat_(storageFormat),
-      bucketProperty_(std::move(bucketProperty)),
-      compressionKind_(compressionKind),
-      serdeParameters_(serdeParameters),
-      writerOptions_(writerOptions),
-      ensureFiles_(ensureFiles),
-      fileNameGenerator_(std::move(fileNameGenerator)),
-      partitionChannels_(computePartitionChannels(inputColumns_)),
-      nonPartitionChannels_(computeNonPartitionChannels(inputColumns_)) {
-  if (compressionKind.has_value()) {
-    VELOX_CHECK(
-        compressionKind.value() != common::CompressionKind_MAX,
-        "Unsupported compression type: CompressionKind_MAX");
-  }
-
-  if (ensureFiles_) {
-    // If ensureFiles is set and either the bucketProperty is set or some
-    // partition keys are in the data, there is not a 1:1 mapping from Task to
-    // files so we can't proactively create writers.
-    VELOX_CHECK(
-        bucketProperty_ == nullptr || bucketProperty_->bucketCount() == 0,
-        "ensureFiles is not supported with bucketing");
-
-    for (const auto& inputColumn : inputColumns_) {
-      VELOX_CHECK(
-          !inputColumn->isPartitionKey(),
-          "ensureFiles is not supported with partition keys in the data");
-    }
-  }
-}
-
 HiveDataSink::HiveDataSink(
     RowTypePtr inputType,
     std::shared_ptr<const HiveInsertTableHandle> insertTableHandle,
@@ -482,13 +443,18 @@ HiveDataSink::HiveDataSink(
                     *insertTableHandle->bucketProperty(),
                     inputType)
               : nullptr,
-          insertTableHandle->partitionChannels(),
-          insertTableHandle->nonPartitionChannels(),
-          createPartitionIdGenerator(
-              inputType,
-              insertTableHandle,
-              hiveConfig,
-              connectorQueryCtx)) {}
+          getPartitionChannels(insertTableHandle),
+          getNonPartitionChannels(insertTableHandle),
+          !getPartitionChannels(insertTableHandle).empty()
+              ? std::make_unique<PartitionIdGenerator>(
+                    inputType,
+                    getPartitionChannels(insertTableHandle),
+                    hiveConfig->maxPartitionsPerWriters(
+                        connectorQueryCtx->sessionProperties()),
+                    connectorQueryCtx->memoryPool(),
+                    hiveConfig->isPartitionPathAsLowerCase(
+                        connectorQueryCtx->sessionProperties()))
+              : nullptr) {}
 
 HiveDataSink::HiveDataSink(
     RowTypePtr inputType,
@@ -592,6 +558,8 @@ void HiveDataSink::appendData(RowVectorPtr input) {
   // Compute partition and bucket numbers.
   computePartitionAndBucketIds(input);
 
+  splitInputRowsAndEnsureWriters(input);
+
   // All inputs belong to a single non-bucketed partition. The partition id
   // must be zero.
   if (!isBucketed() && partitionIdGenerator_->numPartitions() == 1) {
@@ -600,8 +568,6 @@ void HiveDataSink::appendData(RowVectorPtr input) {
     return;
   }
 
-  splitInputRowsAndEnsureWriters();
-
   for (auto index = 0; index < writers_.size(); ++index) {
     const vector_size_t partitionSize = partitionSizes_[index];
     if (partitionSize == 0) {
@@ -1014,7 +980,7 @@ uint32_t HiveDataSink::appendWriter(const HiveWriterId& id) {
 
   std::optional<std::string> partitionName;
   if (isPartitioned()) {
-    partitionName = getPartitionName(id.partitionId.value());
+    partitionName = getPartitionName(id);
   }
 
   // Without explicitly setting flush policy, the default memory based flush
@@ -1082,13 +1048,14 @@ std::unique_ptr<dwio::common::Writer> HiveDataSink::createWriterForIndex(
   return maybeCreateBucketSortWriter(writerIndex, std::move(writer));
 }
 
-std::string HiveDataSink::getPartitionName(uint32_t partitionId) const {
-  VELOX_CHECK_NOT_NULL(partitionIdGenerator_);
-
-  return HivePartitionName::partitionName(
-      partitionId,
-      partitionIdGenerator_->partitionValues(),
-      partitionKeyAsLowerCase_);
+std::optional<std::string> HiveDataSink::getPartitionName(
+    const HiveWriterId& id) const {
+  std::optional<std::string> partitionName;
+  if (isPartitioned()) {
+    partitionName =
+        partitionIdGenerator_->partitionName(id.partitionId.value());
+  }
+  return partitionName;
 }
 
 std::unique_ptr<facebook::velox::dwio::common::Writer>
@@ -1120,6 +1087,13 @@ HiveDataSink::maybeCreateBucketSortWriter(
       sortWriterFinishTimeSliceLimitMs_);
 }
 
+void HiveDataSink::extendBuffersForPartitionedTables() {
+  // Extends the buffer used for partition rows calculations.
+  partitionSizes_.emplace_back(0);
+  partitionRows_.emplace_back(nullptr);
+  rawPartitionRows_.emplace_back(nullptr);
+}
+
 HiveWriterId HiveDataSink::getWriterId(size_t row) const {
   std::optional<int32_t> partitionId;
   if (isPartitioned()) {
@@ -1152,7 +1126,7 @@ void HiveDataSink::updatePartitionRows(
   ++partitionSizes_[index];
 }
 
-void HiveDataSink::splitInputRowsAndEnsureWriters() {
+void HiveDataSink::splitInputRowsAndEnsureWriters(RowVectorPtr /* input */) {
   VELOX_CHECK(isPartitioned() || isBucketed());
   if (isBucketed() && isPartitioned()) {
     VELOX_CHECK_EQ(bucketIds_.size(), partitionIds_.size());
@@ -1165,6 +1139,7 @@ void HiveDataSink::splitInputRowsAndEnsureWriters() {
   for (auto row = 0; row < numRows; ++row) {
     const auto id = getWriterId(row);
     const uint32_t index = ensureWriter(id);
+
     updatePartitionRows(index, numRows, row);
   }