Update file handle

velox/connectors/hive/FileHandle.cpp

@@ -34,7 +34,7 @@ std::string groupName(const std::string& filename) {
 } // namespace
 
 std::shared_ptr<FileHandle> FileHandleGenerator::operator()(
-    const std::string& filename) {
+    const std::tuple<std::string, int64_t, int64_t> fileProperties) {
   // We have seen cases where drivers are stuck when creating file handles.
   // Adding a trace here to spot this more easily in future.
   process::TraceContext trace("FileHandleGenerator::operator()");
@@ -43,12 +43,14 @@ std::shared_ptr<FileHandle> FileHandleGenerator::operator()(
   {
     MicrosecondTimer timer(&elapsedTimeUs);
     fileHandle = std::make_shared<FileHandle>();
-    fileHandle->file = filesystems::getFileSystem(filename, properties_)
-                           ->openFileForRead(filename);
-    fileHandle->uuid = StringIdLease(fileIds(), filename);
-    fileHandle->groupId = StringIdLease(fileIds(), groupName(filename));
-    VLOG(1) << "Generating file handle for: " << filename
-            << " uuid: " << fileHandle->uuid.id();
+    filesystems::FileOptions options;
+    options.fileSize = std::get<1>(fileProperties) == -1 ? std::nullopt
+                                                         : std::optional<int64_t>{std::get<1>(fileProperties)};
+    fileHandle->file = filesystems::getFileSystem(std::get<0>(fileProperties), properties_)
+                           ->openFileForRead(std::get<0>(fileProperties), options);
+    fileHandle->uuid = StringIdLease(fileIds(), std::get<0>(fileProperties));
+    fileHandle->groupId = StringIdLease(fileIds(), groupName(std::get<0>(fileProperties)));
+    VLOG(1) << "Generating file handle for: " << std::get<0>(fileProperties);
   }
   RECORD_HISTOGRAM_METRIC_VALUE(
       kMetricHiveFileHandleGenerateLatencyMs, elapsedTimeUs / 1000);

velox/connectors/hive/FileHandle.h

@@ -54,22 +54,23 @@ struct FileHandle {
   // first diff we'll not include the map.
 };
 
-using FileHandleCache = SimpleLRUCache<std::string, FileHandle>;
+using FileHandleCache = SimpleLRUCache<std::tuple<std::string, int64_t, int64_t>, FileHandle>;
 
 // Creates FileHandles via the Generator interface the CachedFactory requires.
 class FileHandleGenerator {
  public:
   FileHandleGenerator() {}
   FileHandleGenerator(std::shared_ptr<const Config> properties)
       : properties_(std::move(properties)) {}
-  std::shared_ptr<FileHandle> operator()(const std::string& filename);
+  std::shared_ptr<FileHandle> operator()(
+    std::tuple<std::string, int64_t, int64_t> params);
 
  private:
   const std::shared_ptr<const Config> properties_;
 };
 
 using FileHandleFactory = CachedFactory<
-    std::string,
+    std::tuple<std::string, int64_t, int64_t>,
     std::shared_ptr<FileHandle>,
     FileHandleGenerator>;
 

velox/connectors/hive/HiveConnector.cpp

@@ -60,7 +60,7 @@ HiveConnector::HiveConnector(
       fileHandleFactory_(
           hiveConfig_->isFileHandleCacheEnabled()
               ? std::make_unique<
-                    SimpleLRUCache<std::string, std::shared_ptr<FileHandle>>>(
+                    SimpleLRUCache<std::tuple<std::string, int64_t, int64_t>, std::shared_ptr<FileHandle>>>(
                     hiveConfig_->numCacheFileHandles())
               : nullptr,
           std::make_unique<FileHandleGenerator>(config)),

velox/connectors/hive/HiveConnectorSplit.h

@@ -27,6 +27,8 @@ struct HiveConnectorSplit : public connector::ConnectorSplit {
   dwio::common::FileFormat fileFormat;
   const uint64_t start;
   const uint64_t length;
+  const int64_t fileSize;
+  const int64_t modificationTime;
 
   /// Mapping from partition keys to values. Values are specified as strings
   /// formatted the same way as CAST(x as VARCHAR). Null values are specified as
@@ -49,6 +51,8 @@ struct HiveConnectorSplit : public connector::ConnectorSplit {
       dwio::common::FileFormat _fileFormat,
       uint64_t _start = 0,
       uint64_t _length = std::numeric_limits<uint64_t>::max(),
+      int64_t _fileSize = -1,
+      int64_t _modificationTime = -1,
       const std::unordered_map<std::string, std::optional<std::string>>&
           _partitionKeys = {},
       std::optional<int32_t> _tableBucketNumber = std::nullopt,
@@ -62,6 +66,8 @@ struct HiveConnectorSplit : public connector::ConnectorSplit {
         fileFormat(_fileFormat),
         start(_start),
         length(_length),
+        fileSize(_fileSize),
+        modificationTime(_modificationTime),
         partitionKeys(_partitionKeys),
         tableBucketNumber(_tableBucketNumber),
         customSplitInfo(_customSplitInfo),

velox/connectors/hive/SplitReader.cpp

@@ -216,7 +216,8 @@ void SplitReader::createReader() {
 
   std::shared_ptr<FileHandle> fileHandle;
   try {
-    fileHandle = fileHandleFactory_->generate(hiveSplit_->filePath).second;
+    fileHandle = fileHandleFactory_->generate(std::make_tuple(
+      hiveSplit_->filePath, hiveSplit_->fileSize, hiveSplit_->modificationTime)).second;
   } catch (const VeloxRuntimeError& e) {
     if (e.errorCode() == error_code::kFileNotFound &&
         hiveConfig_->ignoreMissingFiles(

velox/connectors/hive/iceberg/IcebergSplit.cpp

@@ -26,6 +26,8 @@ HiveIcebergSplit::HiveIcebergSplit(
     dwio::common::FileFormat _fileFormat,
     uint64_t _start,
     uint64_t _length,
+    int64_t _fileSize,
+    int64_t _modificationTime,
     const std::unordered_map<std::string, std::optional<std::string>>&
         _partitionKeys,
     std::optional<int32_t> _tableBucketNumber,
@@ -38,6 +40,8 @@ HiveIcebergSplit::HiveIcebergSplit(
           _fileFormat,
           _start,
           _length,
+          _fileSize,
+          _modificationTime,
           _partitionKeys,
           _tableBucketNumber,
           _customSplitInfo,
@@ -55,6 +59,8 @@ HiveIcebergSplit::HiveIcebergSplit(
     dwio::common::FileFormat _fileFormat,
     uint64_t _start,
     uint64_t _length,
+    int64_t _fileSize,
+    int64_t _modificationTime,
     const std::unordered_map<std::string, std::optional<std::string>>&
         _partitionKeys,
     std::optional<int32_t> _tableBucketNumber,
@@ -68,6 +74,8 @@ HiveIcebergSplit::HiveIcebergSplit(
           _fileFormat,
           _start,
           _length,
+          _fileSize,
+          _modificationTime,
           _partitionKeys,
           _tableBucketNumber,
           _customSplitInfo,

velox/connectors/hive/iceberg/IcebergSplit.h

@@ -32,6 +32,8 @@ struct HiveIcebergSplit : public connector::hive::HiveConnectorSplit {
       dwio::common::FileFormat _fileFormat,
       uint64_t _start = 0,
       uint64_t _length = std::numeric_limits<uint64_t>::max(),
+      int64_t _fileSize = -1,
+      int64_t _modificationTime = -1,
       const std::unordered_map<std::string, std::optional<std::string>>&
           _partitionKeys = {},
       std::optional<int32_t> _tableBucketNumber = std::nullopt,
@@ -46,6 +48,8 @@ struct HiveIcebergSplit : public connector::hive::HiveConnectorSplit {
       dwio::common::FileFormat _fileFormat,
       uint64_t _start = 0,
       uint64_t _length = std::numeric_limits<uint64_t>::max(),
+      int64_t _fileSize = -1,
+      int64_t _modificationTime = -1,
       const std::unordered_map<std::string, std::optional<std::string>>&
           _partitionKeys = {},
       std::optional<int32_t> _tableBucketNumber = std::nullopt,

velox/connectors/hive/iceberg/PositionalDeleteFileReader.cpp

@@ -94,7 +94,7 @@ PositionalDeleteFileReader::PositionalDeleteFileReader(
       deleteSplit_);
 
   auto deleteFileHandle =
-      fileHandleFactory_->generate(deleteFile_.filePath).second;
+      fileHandleFactory_->generate(make_tuple(deleteFile_.filePath, -1, -1)).second;
   auto deleteFileInput = createBufferedInput(
       *deleteFileHandle,
       deleteReaderOpts,

velox/connectors/hive/iceberg/tests/IcebergReadTest.cpp

@@ -139,6 +139,8 @@ class HiveIcebergTest : public HiveConnectorTestBase {
         fileFomat_,
         0,
         fileSize,
+        -1,
+        -1,
         partitionKeys,
         std::nullopt,
         customSplitInfo,

velox/connectors/hive/storage_adapters/s3fs/tests/S3FileSystemRegistrationTest.cpp

@@ -65,7 +65,7 @@ TEST_F(S3FileSystemRegistrationTest, fileHandle) {
   auto hiveConfig = minioServer_->hiveConfig();
   FileHandleFactory factory(
       std::make_unique<
-          SimpleLRUCache<std::string, std::shared_ptr<FileHandle>>>(1000),
+          SimpleLRUCache<std::tuple<std::string, int64_t, int64_t>, std::shared_ptr<FileHandle>>>(1000),
       std::make_unique<FileHandleGenerator>(hiveConfig));
   auto fileHandle = factory.generate(s3File).second;
   readData(fileHandle->file.get());

velox/connectors/hive/tests/FileHandleTest.cpp

@@ -38,9 +38,9 @@ TEST(FileHandleTest, localFile) {
 
   FileHandleFactory factory(
       std::make_unique<
-          SimpleLRUCache<std::string, std::shared_ptr<FileHandle>>>(1000),
+          SimpleLRUCache<std::tuple<std::string, int64_t, int64_t>, std::shared_ptr<FileHandle>>>(1000),
       std::make_unique<FileHandleGenerator>());
-  auto fileHandle = factory.generate(filename).second;
+  auto fileHandle = factory.generate(std::make_tuple(filename, -1, -1)).second;
   ASSERT_EQ(fileHandle->file->size(), 3);
   char buffer[3];
   ASSERT_EQ(fileHandle->file->pread(0, 3, &buffer), "foo");

velox/connectors/hive/tests/HiveConnectorUtilTest.cpp

@@ -74,6 +74,8 @@ TEST_F(HiveConnectorUtilTest, configureReaderOptions) {
         fileFormat,
         0UL,
         std::numeric_limits<uint64_t>::max(),
+        -1,
+        -1,
         partitionKeys,
         std::nullopt,
         customSplitInfo,

velox/exec/tests/utils/HiveConnectorTestBase.h

@@ -267,6 +267,8 @@ class HiveConnectorSplitBuilder {
         fileFormat_,
         start_,
         length_,
+        fileSize_,
+        modificationTime_,
         partitionKeys_,
         tableBucketNumber_,
         customSplitInfo,
@@ -281,6 +283,8 @@ class HiveConnectorSplitBuilder {
   dwio::common::FileFormat fileFormat_{dwio::common::FileFormat::DWRF};
   uint64_t start_{0};
   uint64_t length_{std::numeric_limits<uint64_t>::max()};
+  int64_t fileSize_{-1};
+  int64_t modificationTime_{-1};
   std::unordered_map<std::string, std::optional<std::string>> partitionKeys_;
   std::optional<int32_t> tableBucketNumber_;
   std::unordered_map<std::string, std::string> customSplitInfo_ = {};