HPCC-36100 Implement directoryFiles for Azure Blob and Files classes

common/remote/hooks/azure/azureapiutils.hpp

@@ -22,6 +22,8 @@
 #include "jlog.hpp"
 #include "jmutex.hpp"
 #include "jstring.hpp"
+#include "jfile.hpp"
+#include "jregexp.hpp"
 
 #include <azure/core.hpp>
 #include <azure/core/http/http.hpp>
@@ -31,6 +33,8 @@
 #include <azure/identity.hpp>
 
 #include <exception>
+#include <optional>
+#include <vector>
 
 /*
  * Common utility functions and constants shared by Azure Blob and File implementations
@@ -52,4 +56,154 @@ void handleRequestException(const Azure::Core::RequestFailedException& e, const
 void handleRequestException(const std::exception& e, const char * op, unsigned attempt, unsigned maxRetries, const char * filename);
 void handleRequestException(const std::exception& e, const char * op, unsigned attempt, unsigned maxRetries, const char * filename, offset_t pos, offset_t len);
 
+struct DirEntry
+{
+    std::string name;
+    bool isDir;
+    int64_t size;
+    time_t modifiedTime;
+};
+
+//---------------------------------------------------------------------------------------------------------------------
+// Base class for Azure directory iterators (Blob and File).
+// Subclasses implement fetchPage() to populate the items vector using the
+// appropriate Azure SDK listing API (with MoveToNextPage() for pagination),
+// resetPaging() to discard stored paged-response state, and createFile() to
+// construct the correct IFile type for each entry.
+
+class AzureDirectoryIteratorBase : implements IDirectoryIterator, public CInterface
+{
+public:
+    IMPLEMENT_IINTERFACE;
+
+    AzureDirectoryIteratorBase(const char *_fullPrefix, const char *_containerOrShare, const char *_mask, bool _includeDirs)
+        : fullPrefix(_fullPrefix), containerOrShare(_containerOrShare), mask(_mask), includeDirs(_includeDirs) {}
+
+    virtual bool first() override
+    {
+        items.clear();
+        itemIndex = 0;
+        hasMorePages = true;
+        resetPaging();
+        return loadNextPage() && advance();
+    }
+    virtual bool next() override    { itemIndex++; return advance(); }
+    virtual bool isValid() override { return curValid; }
+    virtual IFile &query() override
+    {
+        if (!curFile)
+        {
+            StringBuffer fullPath(fullPrefix);
+            fullPath.append(queryCurEntry().name);
+            curFile.setown(createFile(fullPath.str(), items[itemIndex]));
+        }
+        return *curFile;
+    }
+    virtual StringBuffer &getName(StringBuffer &buf) override
+    {
+        if (isValid())
+            buf.append(queryCurEntry().name);
+        return buf;
+    }
+    virtual bool isDir() override { return queryCurEntry().isDir; }
+    virtual __int64 getFileSize() override
+    {
+        const DirEntry &entry = queryCurEntry();
+        return entry.isDir ? -1 : entry.size;
+    }
+    virtual bool getModifiedTime(CDateTime &ret) override
+    {
+        const DirEntry &entry = queryCurEntry();
+        if (entry.isDir)
+        {
+            ret.clear();
+            return false;
+        }
+        ret.set(entry.modifiedTime);
+        return true;
+    }
+
+protected:
+    // Subclass populates 'items' and sets 'hasMorePages' via the SDK's PagedResponse
+    virtual void fetchPage() = 0;
+    // Subclass discards any stored paged-response state so iteration can restart
+    virtual void resetPaging() = 0;
+    // Subclass creates the appropriate IFile (AzureBlob or AzureFile) for the given path
+    virtual IFile *createFile(const char *fullPath, const DirEntry &entry) = 0;
+
+    bool matchesMask(const char *name) const { return mask.isEmpty() || WildMatch(name, mask, false); }
+
+    static time_t toTimeT(const Azure::DateTime &dt)
+    {
+        return std::chrono::system_clock::to_time_t(std::chrono::system_clock::time_point(dt));
+    }
+
+    StringAttr fullPrefix;
+    StringAttr containerOrShare;
+    StringAttr mask;
+    bool includeDirs;
+    std::vector<DirEntry> items;
+    unsigned itemIndex = 0;
+    bool hasMorePages = false;
+
+private:
+    bool loadNextPage()
+    {
+        if (!hasMorePages)
+            return !items.empty();
+        constexpr unsigned maxRetries = 4;
+        unsigned attempt = 0;
+        for (;;)
+        {
+            try
+            {
+                fetchPage();
+                break;
+            }
+            catch (const Azure::Core::RequestFailedException &e)
+            {
+                attempt++;
+                VStringBuffer msg("%s (container: %s)", fullPrefix.str(), containerOrShare.str());
+                handleRequestException(e, "Azure::directoryFiles", attempt, maxRetries, msg.str());
+            }
+            catch (const std::exception &e)
+            {
+                attempt++;
+                VStringBuffer msg("%s (container: %s)", fullPrefix.str(), containerOrShare.str());
+                handleRequestException(e, "Azure::directoryFiles", attempt, maxRetries, msg.str());
+            }
+        }
+        return !items.empty() || hasMorePages;
+    }
+
+    bool advance()
+    {
+        while (itemIndex >= items.size())
+        {
+            if (!hasMorePages)
+            {
+                curFile.clear();
+                curValid = false;
+                return false;
+            }
+            items.clear();
+            itemIndex = 0;
+            if (!loadNextPage())
+            {
+                curFile.clear();
+                curValid = false;
+                return false;
+            }
+        }
+        curFile.clear(); // Created lazily in query() - scanDirectory never calls it
+        curValid = true;
+        return true;
+    }
+
+    inline const DirEntry & queryCurEntry() const { return items[itemIndex]; }
+
+    Owned<IFile> curFile;
+    bool curValid = false;
+};
+
 #endif

common/remote/hooks/azure/azureblob.cpp

@@ -153,10 +153,83 @@ class AzureBlobBlockBlobWriteIO final : implements AzureBlobWriteIO
 };
 
 
+//---------------------------------------------------------------------------------------------------------------------
+// Directory iterator for Azure Blob Storage using ListBlobsByHierarchy API.
+// Uses "/" as a delimiter to simulate directory listing, returning both blobs
+// (files) and blob prefixes (virtual directories) at the current level.
+
+class AzureBlobDirectoryIterator final : public AzureDirectoryIteratorBase
+{
+public:
+    AzureBlobDirectoryIterator(std::shared_ptr<BlobContainerClient> _containerClient,
+                               const char *_prefix, const char *_fullPrefix,
+                               const char *_containerName,
+                               const char *_mask, bool _includeDirs)
+        : AzureDirectoryIteratorBase(_fullPrefix, _containerName, _mask, _includeDirs),
+          containerClient(std::move(_containerClient)), prefix(_prefix) {}
+
+protected:
+    virtual IFile *createFile(const char *fullPath, const DirEntry &entry) override;
+    virtual void resetPaging() override { pagedResponse.reset(); }
+
+    virtual void fetchPage() override
+    {
+        if (!pagedResponse)
+        {
+            ListBlobsOptions options;
+            if (!prefix.empty())
+                options.Prefix = prefix;
+            pagedResponse.emplace(containerClient->ListBlobsByHierarchy("/", options));
+        }
+        else
+        {
+            pagedResponse->MoveToNextPage();
+        }
+        extractItems(*pagedResponse);
+        hasMorePages = pagedResponse->NextPageToken.HasValue() && !pagedResponse->NextPageToken.Value().empty();
+    }
+
+private:
+    void extractItems(const ListBlobsByHierarchyPagedResponse &response)
+    {
+        if (includeDirs)
+        {
+            for (const auto &dp : response.BlobPrefixes)
+            {
+                std::string tail = stripPrefix(dp);
+                if (!tail.empty() && tail.back() == '/')
+                    tail.pop_back();
+                if (!tail.empty() && matchesMask(tail.c_str()))
+                    items.push_back({std::move(tail), true, -1, 0});
+            }
+        }
+        for (const auto &blob : response.Blobs)
+        {
+            std::string tail = stripPrefix(blob.Name);
+            if (!tail.empty() && tail.find('/') == std::string::npos && matchesMask(tail.c_str()))
+                items.push_back({std::move(tail), false, blob.BlobSize, toTimeT(blob.Details.LastModified)});
+        }
+    }
+
+    std::string stripPrefix(const std::string &name) const
+    {
+        if (!prefix.empty() && name.compare(0, prefix.size(), prefix) == 0)
+            return name.substr(prefix.size());
+        return name;
+    }
+
+    std::shared_ptr<BlobContainerClient> containerClient;
+    std::string prefix;
+    std::optional<ListBlobsByHierarchyPagedResponse> pagedResponse;
+};
+
+
 class AzureBlob final : implements CInterfaceOf<IFile>
 {
 public:
     AzureBlob(const char *_azureFileName, AzureAPIConfig &&_config);
+    AzureBlob(const char *_azureFileName, AzureAPIConfig &&_config, const DirEntry &entry)
+    : AzureBlob(_azureFileName, std::move(_config)) { setListedInfo(entry.isDir, entry.size, entry.modifiedTime); }
     virtual bool exists() override
     {
         ensureMetaData();
@@ -211,10 +284,7 @@ class AzureBlob final : implements CInterfaceOf<IFile>
     }
 
 // Directory functions
-    virtual IDirectoryIterator *directoryFiles(const char *mask, bool sub, bool includeDirs) override
-    {
-        UNIMPLEMENTED_X("AzureBlob::directoryFiles");
-    }
+    virtual IDirectoryIterator *directoryFiles(const char *mask, bool sub, bool includeDirs) override;
     virtual bool getInfo(bool &isdir,offset_t &size,CDateTime &modtime) override
     {
         ensureMetaData();
@@ -259,6 +329,15 @@ class AzureBlob final : implements CInterfaceOf<IFile>
 public:
     SharedBlobClient getBlobClient() const;
     void invalidateMeta() { haveMeta = false; }
+    void setListedInfo(bool _isDir, offset_t _fileSize, time_t _lastModified)
+    {
+        haveMeta = true;
+        isDir = _isDir;
+        fileExists = true;
+        fileSize = _fileSize;
+        lastModified = _lastModified;
+        createdOn = 0;
+    }
     const AzureAPIConfig & queryConfig() const { return config; }
 
 protected:
@@ -624,10 +703,10 @@ AzureBlob::AzureBlob(const char *_azureFileName, AzureAPIConfig &&_config) : ful
             if ((device == 0) || (device > numDevices))
                 throw makeStringExceptionV(99, "Device %d out of range for plane %s", device, planeName.str());
 
-            if (!endDevice || (*endDevice != '/'))
+            if (!endDevice || (*endDevice != '/' && *endDevice != '\0'))
                 throw makeStringExceptionV(99, "Unexpected end of device partition %s", fullName.str());
 
-            filename = endDevice+1;
+            filename = (*endDevice == '/') ? endDevice+1 : endDevice;
         }
 
         VStringBuffer childPath("containers[%u]", device);
@@ -660,10 +739,14 @@ AzureBlob::AzureBlob(const char *_azureFileName, AzureAPIConfig &&_config) : ful
     }
     else
         throw makeStringExceptionV(99, "Unexpected prefix on azure filename %s", fullName.str());
-
     blobUrl = ::getBlobUrl(accountName, containerName, blobName);
 }
 
+IFile * AzureBlobDirectoryIterator::createFile(const char *fullPath, const DirEntry &entry)
+{
+    return createAzureBlob(fullPath, entry);
+}
+
 std::shared_ptr<StorageSharedKeyCredential> AzureBlob::getSharedKeyCredentials() const
 {
     return getAzureSharedKeyCredential(accountName.str(), secretName.str());
@@ -786,6 +869,27 @@ IFileIO * AzureBlob::createFileWriteIO()
     return new AzureBlobBlockBlobWriteIO(this);
 }
 
+IDirectoryIterator * AzureBlob::directoryFiles(const char *mask, bool sub, bool includeDirs)
+{
+    if (sub)
+        UNIMPLEMENTED_X("AzureBlob::directoryFiles recursive");
+
+    // Build the blob prefix for listing. The blobName is the path within the container.
+    // For directory listing, we want to list blobs under this prefix with "/" delimiter.
+    std::string listPrefix(blobName.str());
+    if (!listPrefix.empty() && listPrefix.back() != '/')
+        listPrefix += '/';
+
+    // Build the full azure path prefix for constructing child IFile paths
+    // fullName is e.g. "azureblob:plane/path" or "azureblob:plane/d1/path"
+    StringBuffer fullPathPrefix(fullName);
+    if (fullPathPrefix.length() > 0 && fullPathPrefix.charAt(fullPathPrefix.length()-1) != '/')
+        fullPathPrefix.append('/');
+
+    auto containerClient = getBlobContainerClient();
+    return new AzureBlobDirectoryIterator(containerClient, listPrefix.c_str(), fullPathPrefix.str(), containerName.str(), mask, includeDirs);
+}
+
 void AzureBlob::ensureMetaData()
 {
     CriticalBlock block(cs);
@@ -903,3 +1007,8 @@ IFile *createAzureBlob(const char *azureFileName)
 {
     return new AzureBlob(azureFileName, getAzureConfig());
 }
+
+IFile *createAzureBlob(const char *azureFileName, const DirEntry &entry)
+{
+    return new AzureBlob(azureFileName, getAzureConfig(), entry);
+}

common/remote/hooks/azure/azureblob.hpp

@@ -19,6 +19,7 @@
 #define AZURE_BLOB_HPP
 
 #include "jfile.hpp"
+#include "azureapiutils.hpp"
 
 /*
  * Direct access to files in Azure blobs
@@ -27,5 +28,6 @@
  */
 
 IFile * createAzureBlob(const char * filename);
+IFile * createAzureBlob(const char * filename, const DirEntry &entry);
 
 #endif