|| || || || || ||

---

Database.h

Go to the documentation of this file.

#ifndef Impala_Util_Database_h
#define Impala_Util_Database_h

#ifndef REPOSITORY_USED // Here comes the deprecated stuff

#ifdef unix
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/resource.h>
#else
#include <direct.h>
#include <errno.h>
#endif

#include <vector>

#include "Basis/CmdOptions.h"
#include "Basis/FileNameTmp.h"
#include "Util/ChannelPool.h"
#include "Util/ReadIOBufferFromChannel.h"
#include "Util/IOBufferFile.h"
#include "Util/IOBufferChannel.h"

namespace Impala
{
namespace Util
{


class Database
{
public:

    static Database&
    GetInstance()
    {
        static Database theDatabase;
        return theDatabase;
    }

    Database()
    {
        Init("");
    }

    Database(String setName)
    {
        Init(setName);
    }

    // For use in data server only : reset data path for new request.
    void
    SetDataPath(String data)
    {
        mDataString = data;
        mData.clear();
        mData = StringList(mDataString, ';');
    }

    String
    GetDataPath()
    {
        return mDataString;
    }

    // For use in data server only : reset override for new request.
    void
    SetOverride(bool override)
    {
        mOverride = override;
    }

    // Any data directory part

    String
    MakeDir(CString dir)
    {
        return DoMkDirPath(dir);
    }

    String
    MakeDir(CString dir1, CString dir2)
    {
        return MakeDir(ConcatDir(dir1, dir2));
    }

    String
    MakeDir(CString dir1, CString dir2, CString dir3)
    {
        return MakeDir(dir1, ConcatDir(dir2, dir3));
    }

    String
    MakeDir(CString dir1, CString dir2, CString dir3, CString dir4)
    {
        return MakeDir(dir1, dir2, ConcatDir(dir3, dir4));
    }

    String
    MakeDir(CString dir1, CString dir2, CString dir3, CString dir4, CString dir5)
    {
        return MakeDir(dir1, dir2, dir3, ConcatDir(dir4, dir5));
    }

    // Any data file part

    String
    GetFilePath(CString file, bool toWrite, bool silent)
    {
        CString path = file;
        if (toWrite)
            return GetWritableFile(mData.begin(), mData.end(), path, silent);
        return GetReadableFile(mData.begin(), mData.end(), path, silent);
    }

    String
    GetFilePath(CString dir, CString file, bool toWrite, bool silent)
    {
        String path = dir + "/" + file;
        if (toWrite)
            return GetWritableFile(mData.begin(), mData.end(), path, silent);
        return GetReadableFile(mData.begin(), mData.end(), path, silent);
    }

    String
    GetFilePath(CString dir1, CString dir2, CString file, bool toWrite,
                bool silent)
    {
        String path = dir1 + "/" + dir2 + "/" + file;
        if (toWrite)
            return GetWritableFile(mData.begin(), mData.end(), path, silent);
        return GetReadableFile(mData.begin(), mData.end(), path, silent);
    }

    String
    GetFilePath(CString dir1, CString dir2, CString dir3, CString file,
                bool toWrite, bool silent)
    {
        String path = dir1 + "/" + dir2 + "/" + dir3 + "/" + file;
        if (toWrite)
            return GetWritableFile(mData.begin(), mData.end(), path, silent);
        return GetReadableFile(mData.begin(), mData.end(), path, silent);
    }

    String
    GetFilePath(CString dir1, CString dir2, CString dir3, CString dir4,
                CString file, bool toWrite, bool silent)
    {
        String path = dir1 + "/" + dir2 + "/" + dir3 + "/" + dir4 + "/" + file;
        if (toWrite)
            return GetWritableFile(mData.begin(), mData.end(), path, silent);
        return GetReadableFile(mData.begin(), mData.end(), path, silent);
    }

    String
    GetFilePath(CString dir1, CString dir2, CString dir3, CString dir4,
                CString dir5, CString file, bool toWrite, bool silent)
    {
        String path = dir1 + "/" + dir2 + "/" + dir3 + "/" + dir4 + "/"
            + dir5 + "/" + file;
        if (toWrite)
            return GetWritableFile(mData.begin(), mData.end(), path, silent);
        return GetReadableFile(mData.begin(), mData.end(), path, silent);
    }

    void
    DeleteFile(CString path)
    {
        if (mDataChannel)
        {
            char* buf = mDataChannel->Buffer();
            sprintf(buf, "deletefile \"%s\"\0", path.c_str());
            mDataChannel->SendRequest(strlen(buf)+1);
        }
        else
        {
            unlink(path.c_str());
        }
    }

    IOBuffer*
    GetIOBuffer(CString fileName, bool readMode, bool useMemoryIfLocal,
                String useLocalFileIfRemote, Int64 useLocalSizeIfRemote = 0,
                bool useIOBufferChannel = false)
    {
        IOBuffer* res = 0;
        ILOG_DEBUG("GetIOBuffer [" << fileName << "]");
        if (mDataChannel)
        {
            if (readMode)
            {
                if (useIOBufferChannel)
                {
                    ILOG_DEBUG("GetIOBuffer: Read using Channel to " <<
                               mDataServer << ": " << fileName);
                    res = new IOBufferChannel(fileName, true, mDataChannel);
                }
                else
                {
                    ILOG_DEBUG("GetIOBuffer: Read using File: " << fileName);
                    /* the race condition that occurs when someone passes in
                       "tmp" while reading a file is very difficult to track
                       down (only happens with MPI-jobs). Therefore, generate 
                       a temporary filename even in read mode */
                    if (useLocalFileIfRemote == "tmp")
                        useLocalFileIfRemote = FileNameTmp();
                    res = ReadIOBufferFromChannel(mDataChannel, fileName,
                                                  useLocalFileIfRemote);
                }
            }
            else
            {
                if (useIOBufferChannel)
                {
                    ILOG_DEBUG("GetIOBuffer: Write using Channel " << fileName);
                    res = new IOBufferChannel(fileName, false, mDataChannel);
                }
                else if (useLocalFileIfRemote.empty())
                {
                    ILOG_DEBUG("GetIOBuffer: Write using mem " << fileName);
                    res = new IOBuffer(useLocalSizeIfRemote);
                }
                else
                {
                    ILOG_DEBUG("GetIOBuffer: Write using File " << fileName);
                    if (useLocalFileIfRemote == "tmp")
                        useLocalFileIfRemote = FileNameTmp();
                    res = new IOBufferFile(useLocalFileIfRemote, false, false);
                }
                if (!useIOBufferChannel)
                    res->SetUseChannel(mDataChannel, fileName,
                                       useLocalFileIfRemote);
            }
        }
        else
        {
            res = new IOBufferFile(fileName, readMode, useMemoryIfLocal);
        }

        if (res)
        {
            if (!res->Valid())
            {
                delete res;
                res = 0;
            }
        }
        return res;
    }

    Channel*
    GetDataChannel()
    {
        return mDataChannel;
    }

    void
    Dump() const
    {
        ILOG_INFO_HEADNODE("Dump: data=[" << mDataString << "] server=[" <<
                       mDataServer << "]");
    }

private:

    void
    Init(String setName)
    {
        CmdOptions& options = CmdOptions::GetInstance();
        mDataString = StringReplaceAll(options.GetString("data"), ":", ";", false);
        ILOG_DEBUG("'data' option externally specified: " << mDataString);
        String additionalData = options.GetString("data::"+setName, "");
        if (additionalData.empty() && !setName.empty())
            additionalData = "../" + FileNameBase(setName);
        if (!additionalData.empty())
        {
            mDataString = additionalData + ";" + mDataString;
            ILOG_INFO_HEADNODE("'data' option for " << setName <<
                           " has been extended: " << mDataString);
        }
        mData = StringList(mDataString, ';');
        mDataServer = options.GetString("dataServer");
        String replaceDataServer = options.GetString("dataServer::"+setName, "");
        if (!replaceDataServer.empty())
        {
            mDataServer = replaceDataServer;
            ILOG_INFO_HEADNODE("server for " << setName << " is " << mDataServer);
        }
        mDataChannel = 0;
        if (!mDataServer.empty())
        {
            ILOG_INFO_HEADNODE("Retrieving channel to server " << mDataServer);
            String passFile = options.GetString("passwordFile");
            mDataChannel = ChannelPool::Instance().Get(mDataServer, passFile);
        }
        mOverride = options.GetBool("override");
#ifdef WIN32
        // printf("%d\n", _getmaxstdio()); // normally 512
        _setmaxstdio(2048); // set it to the absolute maximum
#endif
#ifdef unix
        rlimit lim;
        getrlimit(RLIMIT_NOFILE, &lim);
        lim.rlim_cur = 2048; // will succeed on image, down to 1024(?) on linux
        setrlimit(RLIMIT_NOFILE, &lim);
#endif
    }

    // Utilities

    void
    DoMkDir(CString dir)
    {
#ifdef unix
        mode_t m = 0755;
        mkdir(dir.c_str(), m);
#else
        if (_mkdir(StringReplaceAll(dir, "/", "\\").c_str()) != 0 && 
            errno != EEXIST)
            ILOG_ERROR("[DoMkDir] Failed to create directory: " << dir << 
                       " (" << strerror(errno) << ")");
#endif
    }

    String
    DoMkDirPath(CString dir)
    {
        if (mDataChannel)
            return DoMkDirPathServer(dir);

        return DoMkDirPathLocal(ConcatDir(*mData.begin(), dir));
    }

    String
    DoMkDirPathLocal(CString dirWithBackSlashes)
    {
        String dir = StringReplaceAll(dirWithBackSlashes, "\\", "/");
        ILOG_DEBUG("DoMkDirPath: " << dir);

        String path = "";
        if (dir[0] == '/')
        {
#ifdef WIN32
            if (dir.find("//") == 0)
            {
                // construct path head upto and including fourth slash
                int headEndPos = 1;
                int nextSlashPos = dir.find("/", headEndPos + 1);
                if (nextSlashPos != String::npos)
                {
                    headEndPos = nextSlashPos;
                    nextSlashPos = dir.find("/", headEndPos + 1);
                    if (nextSlashPos != String::npos)
                        headEndPos = nextSlashPos;
                }
                path = dir.substr(0, headEndPos + 1);
                dir = dir.substr(headEndPos + 1);
            }
            else
#endif
                path = String("/");
        }
        
        StringList subDirs(dir, '/');
        for (StringListCI i=subDirs.begin() ; i!=subDirs.end() ; i++)
        {
            CString subDir = *i;
            ILOG_DEBUG("  DoMkDirPath: subDir = " << subDir);
            if (subDir.size() == 0 || subDir == ".")
                continue;
            if ((path.size() > 0) && (*path.rbegin() != '/'))
                path = path + "/";
            path = path + *i;
            DoMkDir(path);
        }
        return path;
    }

    String
    DoMkDirPathServer(CString dir)
    {
        char* buf = mDataChannel->Buffer();
        sprintf(buf, "domkdirpath \"%s\" \"%s\"\0", dir.c_str(), mDataString.c_str());
        int len = mDataChannel->SendRequest(strlen(buf)+1);
        if (mDataChannel->LastSendHadError())
            return "";
        buf[len] = 0;
        String res(buf);
        ILOG_DEBUG("Got dir [" << res << "]");
        return res;
    }

    String
    GetReadableFile(StringListCI begin, StringListCI end, CString file,
                    bool silent)
    {
        if (mDataChannel)
            return GetReadableFileServer(file, silent);
        return GetReadableFileLocal(begin, end, file, silent);
    }

    String
    GetReadableFileLocal(StringListCI begin, StringListCI end, CString file,
                         bool silent)
    {
        StringListCI i;
        for (i=begin ; i!=end ; i++)
        {
            String d = StringResolveEnv(*i);
            String path = FileNameConcat(d, file);
            ILOG_DEBUG("GetReadableFile: trying " << path);
            if (IOBufferFile::FileExists(path))
                return path;
        }

        if (!silent)
        {
            ILOG_ERROR("Unable to find " << file << " in path ");
            for (i=begin ; i!=end ; i++)
                ILOG_INFO("    " << *i);
        }
        return String("");
    }

    String
    GetReadableFileServer(CString file, bool silent)
    {
        char* buf = mDataChannel->Buffer();
        sprintf(buf, "getreadablefile \"%s\" \"%s\" %d\0", file.c_str(),
                mDataString.c_str(), silent);
        int len = mDataChannel->SendRequest(strlen(buf)+1);
        if (mDataChannel->LastSendHadError())
            return "";
        buf[len] = 0;
        String res(buf);
        ILOG_DEBUG("Got file [" << res << "] from " << mDataServer);
        return res;
    }

    String
    GetWritableFile(StringListCI begin, StringListCI end, CString file,
                    bool silent)
    {
        if (mDataChannel)
            return GetWritableFileServer(file, silent);
        return GetWritableFileLocal(begin, end, file, silent);
    }

    String
    GetWritableFileLocal(StringListCI begin, StringListCI end, CString file,
                         bool silent)
    {
        if (begin == end)
        {
            ILOG_ERROR("GetWritableFile: no search path");
            return String("");
        }
        String d = StringResolveEnv(*begin);
        String path = FileNameConcat(d, file);
        ILOG_DEBUG("GetWritableFile: trying " << path);
        if (IOBufferFile::FileExists(path))
        {
            if (! (mOverride || silent))   // silent implies "just checking"
            {
                if (!silent)
                    ILOG_ERROR(path << " already exists, will not override!");
                return String("");
            }
        }
        return path;
    }

    String
    GetWritableFileServer(CString file, bool silent)
    {
        char* buf = mDataChannel->Buffer();
        sprintf(buf, "getwritablefile \"%s\" \"%s\" %d %d\0", file.c_str(),
                mDataString.c_str(), silent, mOverride);
        int len = mDataChannel->SendRequest(strlen(buf)+1);
        if (mDataChannel->LastSendHadError())
            return "";
        buf[len] = 0;
        String res(buf);
        ILOG_DEBUG("Got file [" << res << "] from " << mDataServer);
        return res;
    }

    String ConcatDir(CString dir, CString subDir)
    {
        return (subDir.empty() ? dir : (dir.empty() ? subDir : (dir + "/" + subDir)));
    }

    bool        mOverride;
    String      mDataString;
    StringList  mData;
    String      mDataServer;
    Channel*    mDataChannel;

    ILOG_VAR_DEC;
};

ILOG_VAR_INIT(Database, Impala.Util);

} // namespace Util
} // namespace Impala

#endif // REPOSITORY_USED

#endif

---