[Native] Allow '$file_size' and '$file_modified_time' for HiveSplits in SQL queries

presto-native-execution/presto_cpp/main/types/PrestoToVeloxQueryPlan.cpp

@@ -948,12 +948,28 @@ std::shared_ptr<connector::ConnectorTableHandle> toConnectorTableHandle(
     const VeloxExprConverter& exprConverter,
     const TypeParser& typeParser,
     std::unordered_map<std::string, std::shared_ptr<connector::ColumnHandle>>&
-        partitionColumns) {
+        assignments) {
+  auto addSynthesizedColumn = [&](const std::string& name,
+                                  protocol::ColumnType columnType,
+                                  const protocol::ColumnHandle& column) {
+    if (toHiveColumnType(columnType) ==
+        velox::connector::hive::HiveColumnHandle::ColumnType::kSynthesized) {
+      if (assignments.count(name) == 0) {
+        assignments.emplace(name, toColumnHandle(&column, typeParser));
+      }
+    }
+  };
+
   if (auto hiveLayout =
           std::dynamic_pointer_cast<const protocol::HiveTableLayoutHandle>(
               tableHandle.connectorTableLayout)) {
     for (const auto& entry : hiveLayout->partitionColumns) {
-      partitionColumns.emplace(entry.name, toColumnHandle(&entry, typeParser));
+      assignments.emplace(entry.name, toColumnHandle(&entry, typeParser));
+    }
+
+    // Add synthesized columns to the TableScanNode columnHandles as well.
+    for (const auto& entry : hiveLayout->predicateColumns) {
+      addSynthesizedColumn(entry.first, entry.second.columnType, entry.second);
     }
 
     auto hiveTableHandle =
@@ -983,10 +999,15 @@ std::shared_ptr<connector::ConnectorTableHandle> toConnectorTableHandle(
           std::dynamic_pointer_cast<const protocol::IcebergTableLayoutHandle>(
               tableHandle.connectorTableLayout)) {
     for (const auto& entry : icebergLayout->partitionColumns) {
-      partitionColumns.emplace(
+      assignments.emplace(
           entry.columnIdentity.name, toColumnHandle(&entry, typeParser));
     }
 
+    // Add synthesized columns to the TableScanNode columnHandles as well.
+    for (const auto& entry : icebergLayout->predicateColumns) {
+      addSynthesizedColumn(entry.first, entry.second.columnType, entry.second);
+    }
+
     auto icebergTableHandle =
         std::dynamic_pointer_cast<const protocol::IcebergTableHandle>(
             tableHandle.connectorHandle);

presto-native-execution/presto_cpp/main/types/PrestoToVeloxSplit.cpp

@@ -97,6 +97,15 @@ velox::exec::Split toVeloxSplit(
     for (const auto& [key, value] : hiveSplit->storage.serdeParameters) {
       serdeParameters[key] = value;
     }
+
+    std::unordered_map<std::string, std::string> infoColumns;
+    infoColumns.reserve(2);
+    infoColumns.insert(
+        {"$file_size", std::to_string(hiveSplit->fileSplit.fileSize)});
+    infoColumns.insert(
+        {"$file_modified_time",
+         std::to_string(hiveSplit->fileSplit.fileModifiedTime)});
+
     return velox::exec::Split(
         std::make_shared<connector::hive::HiveConnectorSplit>(
             scheduledSplit.split.connectorId,
@@ -111,7 +120,8 @@ velox::exec::Split toVeloxSplit(
             customSplitInfo,
             extraFileInfo,
             serdeParameters,
-            hiveSplit->splitWeight),
+            hiveSplit->splitWeight,
+            infoColumns),
         splitGroupId);
   }
 

presto-native-execution/src/test/java/com/facebook/presto/nativeworker/AbstractTestNativeGeneralQueries.java

@@ -955,8 +955,19 @@ public void testPath()
 
         // Fetch one of the file paths and use it in a filter
         String path = (String) computeActual("SELECT \"$path\" from orders LIMIT 1").getOnlyValue();
-
         assertQuery(format("SELECT * from orders WHERE \"$path\"='%s'", path));
+
+        assertQuery("SELECT \"$file_size\", * from orders");
+
+        // Fetch one of the file sizes and use it in a filter
+        Long fileSize = (Long) computeActual("SELECT \"$file_size\" from orders LIMIT 1").getOnlyValue();
+        assertQuery(format("SELECT * from orders WHERE \"$file_size\"=%d", fileSize));
+
+        assertQuery("SELECT \"$file_modified_time\", * from orders");
+
+        // Fetch one of the file modified times and use it as a filter.
+        Long fileModifiedTime = (Long) computeActual("SELECT \"$file_modified_time\" from orders LIMIT 1").getOnlyValue();
+        assertQuery(format("SELECT *, \"$file_modified_time\" from orders WHERE \"$file_modified_time\"=%d", fileModifiedTime));
     }
 
     @Test