FileStorageManager extends StorageManager { public FileSystem getFileSystem() { return this.fs; } FileStorageManager(String storeType); Scanner getFileScanner(TableMeta meta, Schema schema, Path path); Scanner getFileScanner(TableMeta meta, Schema schema, Path path, FileStatus status); FileSystem getFileSystem(); Path getWarehouseDir(); void delete(Path tablePath); boolean exists(Path path); void deleteData(Path path); Path getTablePath(String tableName); void setPartitionPath(String path); void setCurrentDepth(int depth); @VisibleForTesting Appender getAppender(TableMeta meta, Schema schema, Path filePath); FileFragment[] split(String tableName); FileFragment[] split(String tableName, long fragmentSize); FileFragment[] splitBroadcastTable(Path tablePath); FileFragment[] split(Path tablePath); FileFragment[] split(String tableName, Path tablePath); static FileFragment[] splitNG(Configuration conf, String tableName, TableMeta meta,
                                       Path tablePath, long size); long calculateSize(Path tablePath); Path getAppenderFilePath(TaskAttemptId taskAttemptId, Path workDir); long getMinSplitSize(); List<Fragment> getSplits(String tableName, TableMeta meta, Schema schema, Path... inputs); @Override List<Fragment> getSplits(String tableName, TableDesc table, ScanNode scanNode); @Override void createTable(TableDesc tableDesc, boolean ifNotExists); @Override void purgeTable(TableDesc tableDesc); @Override List<Fragment> getNonForwardSplit(TableDesc tableDesc, int currentPage, int numResultFragments); @Override StorageProperty getStorageProperty(); @Override void closeStorageManager(); @Override void beforeInsertOrCATS(LogicalNode node); @Override void rollbackOutputCommit(LogicalNode node); @Override TupleRange[] getInsertSortRanges(OverridableConf queryContext, TableDesc tableDesc,
                                          Schema inputSchema, SortSpec[] sortSpecs, TupleRange dataRange); static synchronized SeekableScanner getSeekableScanner(
      TajoConf conf, TableMeta meta, Schema schema, Path path);  }
@Test public void testStoreType() throws Exception { final Configuration hdfsConf = new HdfsConfiguration(); String testDataPath = TEST_PATH + "/" + UUID.randomUUID().toString(); hdfsConf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, testDataPath); hdfsConf.setLong(DFSConfigKeys.DFS_NAMENODE_MIN_BLOCK_SIZE_KEY, 0); hdfsConf.setBoolean(DFSConfigKeys.DFS_HDFS_BLOCKS_METADATA_ENABLED, true); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(hdfsConf) .numDataNodes(2).build(); cluster.waitClusterUp(); TajoConf tajoConf = new TajoConf(hdfsConf); tajoConf.setVar(TajoConf.ConfVars.ROOT_DIR, cluster.getFileSystem().getUri() + "/tajo"); try { FileStorageManager sm = (FileStorageManager)StorageManager.getStorageManager(conf, "CSV"); assertEquals(fs.getUri(), sm.getFileSystem().getUri()); sm = (FileStorageManager)StorageManager.getStorageManager(tajoConf, "CSV"); assertNotEquals(fs.getUri(), sm.getFileSystem().getUri()); assertEquals(cluster.getFileSystem().getUri(), sm.getFileSystem().getUri()); } finally { cluster.shutdown(true); } }