feat(filestore): support reading nested directory and various file

core/src/main/java/cn/edu/tsinghua/iginx/engine/physical/memory/execute/stream/EmptyRowStream.java

@@ -23,13 +23,18 @@
 import cn.edu.tsinghua.iginx.engine.shared.data.read.Row;
 import cn.edu.tsinghua.iginx.engine.shared.data.read.RowStream;
 import java.util.Collections;
+import java.util.Objects;
 
 public class EmptyRowStream implements RowStream {
 
   private final Header header;
 
   public EmptyRowStream() {
-    this.header = new Header(Field.KEY, Collections.emptyList());
+    this(new Header(Field.KEY, Collections.emptyList()));
+  }
+
+  public EmptyRowStream(Header header) {
+    this.header = Objects.requireNonNull(header);
   }
 
   @Override

dataSource/filestore/src/main/java/cn/edu/tsinghua/iginx/filestore/FileStorage.java

@@ -38,6 +38,7 @@
 import cn.edu.tsinghua.iginx.engine.shared.operator.filter.Filter;
 import cn.edu.tsinghua.iginx.engine.shared.operator.tag.TagFilter;
 import cn.edu.tsinghua.iginx.filestore.common.AbstractConfig;
+import cn.edu.tsinghua.iginx.filestore.common.Configs;
 import cn.edu.tsinghua.iginx.filestore.common.FileStoreException;
 import cn.edu.tsinghua.iginx.filestore.common.Filters;
 import cn.edu.tsinghua.iginx.filestore.service.FileStoreConfig;
@@ -48,6 +49,7 @@
 import cn.edu.tsinghua.iginx.filestore.struct.FileStructureManager;
 import cn.edu.tsinghua.iginx.filestore.struct.legacy.filesystem.LegacyFilesystem;
 import cn.edu.tsinghua.iginx.filestore.struct.legacy.parquet.LegacyParquet;
+import cn.edu.tsinghua.iginx.filestore.struct.tree.FileTreeConfig;
 import cn.edu.tsinghua.iginx.filestore.thrift.DataBoundary;
 import cn.edu.tsinghua.iginx.filestore.thrift.DataUnit;
 import cn.edu.tsinghua.iginx.metadata.entity.ColumnsInterval;
@@ -56,8 +58,8 @@
 import cn.edu.tsinghua.iginx.thrift.AggregateType;
 import cn.edu.tsinghua.iginx.thrift.StorageEngineType;
 import cn.edu.tsinghua.iginx.utils.Pair;
+import com.google.common.base.Strings;
 import com.typesafe.config.Config;
-import com.typesafe.config.ConfigBeanFactory;
 import com.typesafe.config.ConfigFactory;
 import java.net.InetSocketAddress;
 import java.util.*;
@@ -102,7 +104,7 @@ static FileStoreConfig toFileStoreConfig(StorageEngineMeta meta)
       throws StorageInitializationException {
     Config rawConfig = toConfig(meta);
     LOGGER.debug("storage of {} config: {}", meta, rawConfig);
-    FileStoreConfig fileStoreConfig = ConfigBeanFactory.create(rawConfig, FileStoreConfig.class);
+    FileStoreConfig fileStoreConfig = FileStoreConfig.of(rawConfig);
     LOGGER.debug("storage of {} will be initialized with {}", meta, fileStoreConfig);
     List<AbstractConfig.ValidationProblem> problems = fileStoreConfig.validate();
     if (!problems.isEmpty()) {
@@ -112,26 +114,41 @@ static FileStoreConfig toFileStoreConfig(StorageEngineMeta meta)
   }
 
   static Config toConfig(StorageEngineMeta meta) throws StorageInitializationException {
-    HashMap<String, String> reshapedParams = new HashMap<>();
+    HashMap<String, String> reshaped = new HashMap<>();
 
     for (Map.Entry<String, String> param : meta.getExtraParams().entrySet()) {
       String key = param.getKey();
       String value = param.getValue();
       if (key.contains(".")) {
-        reshapedParams.put(key, value);
+        reshaped.put(key, value);
       }
     }
 
-    reshapedParams.put("data.root", meta.getExtraParams().get("dir"));
-    reshapedParams.put("dummy.root", meta.getExtraParams().get("dummy_dir"));
-    reshapedParams.putIfAbsent("data.struct", LegacyParquet.NAME);
-    reshapedParams.putIfAbsent("dummy.struct", LegacyFilesystem.NAME);
+    Configs.put(
+        reshaped,
+        meta.getExtraParams().get("dir"),
+        FileStoreConfig.Fields.data,
+        StorageConfig.Fields.root);
+    Configs.put(
+        reshaped,
+        meta.getExtraParams().get("dummy_dir"),
+        FileStoreConfig.Fields.dummy,
+        StorageConfig.Fields.root);
+    Configs.put(
+        reshaped,
+        meta.getExtraParams().get("embedded_prefix"),
+        FileStoreConfig.Fields.dummy,
+        StorageConfig.Fields.config,
+        FileTreeConfig.Fields.prefix);
+    Configs.putIfAbsent(
+        reshaped, LegacyParquet.NAME, FileStoreConfig.Fields.data, StorageConfig.Fields.struct);
+    Configs.putIfAbsent(
+        reshaped, LegacyFilesystem.NAME, FileStoreConfig.Fields.dummy, StorageConfig.Fields.struct);
 
     boolean local = isLocal(meta);
-    reshapedParams.put("server", String.valueOf(local));
+    reshaped.put(FileStoreConfig.Fields.serve, String.valueOf(local));
 
-    Config config =
-        ConfigFactory.parseMap(reshapedParams, "storage engine initialization parameters");
+    Config config = ConfigFactory.parseMap(reshaped, "storage engine initialization parameters");
 
     if (local) {
       LOGGER.debug("storage of {} is local, ignore config for remote", meta);
@@ -308,7 +325,7 @@ public List<Column> getColumns(Set<String> patterns, TagFilter tagFilter)
   @Override
   public Pair<ColumnsInterval, KeyInterval> getBoundaryOfStorage(String prefix)
       throws PhysicalException {
-    Map<DataUnit, DataBoundary> units = service.getUnits(prefix);
+    Map<DataUnit, DataBoundary> units = service.getUnits(Strings.emptyToNull(prefix));
     DataBoundary boundary = units.get(unitOfDummy());
     if (Objects.equals(boundary, new DataBoundary())) {
       throw new PhysicalTaskExecuteFailureException("no data");

dataSource/filestore/src/main/java/cn/edu/tsinghua/iginx/filestore/common/AbstractConfig.java

@@ -18,6 +18,8 @@
 package cn.edu.tsinghua.iginx.filestore.common;
 
 import com.google.common.collect.Range;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigBeanFactory;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -29,6 +31,10 @@ public abstract class AbstractConfig {
 
   public abstract List<ValidationProblem> validate();
 
+  public static <C extends AbstractConfig> C of(Config raw, Class<C> clazz) {
+    return ConfigBeanFactory.create(raw, clazz);
+  }
+
   public static class ValidationProblem {
     private final List<String> reversedPath;
     private final String problem;

dataSource/filestore/src/main/java/cn/edu/tsinghua/iginx/filestore/common/Closeables.java

@@ -0,0 +1,65 @@
+/*
+ * IGinX - the polystore system with high performance
+ * Copyright (C) Tsinghua University
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package cn.edu.tsinghua.iginx.filestore.common;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+public class Closeables {
+
+  private Closeables() {}
+
+  public static void close(Iterable<? extends Closeable> ac) throws IOException {
+    if (ac == null) {
+      return;
+    } else if (ac instanceof Closeable) {
+      ((Closeable) ac).close();
+      return;
+    }
+
+    IOException exception = null;
+    for (Closeable closeable : ac) {
+      try {
+        if (closeable != null) {
+          closeable.close();
+        }
+      } catch (IOException e) {
+        if (exception == null) {
+          exception = e;
+        } else if (e != exception) {
+          exception.addSuppressed(e);
+        }
+      }
+    }
+    if (exception != null) {
+      throw exception;
+    }
+  }
+
+  public static Closeable closeAsIOException(AutoCloseable ac) {
+    return () -> {
+      try {
+        ac.close();
+      } catch (RuntimeException e) {
+        throw e;
+      } catch (Exception e) {
+        throw new IOException(e);
+      }
+    };
+  }
+}

dataSource/filestore/src/main/java/cn/edu/tsinghua/iginx/filestore/common/Configs.java

@@ -0,0 +1,40 @@
+/*
+ * IGinX - the polystore system with high performance
+ * Copyright (C) Tsinghua University
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package cn.edu.tsinghua.iginx.filestore.common;
+
+import java.util.Map;
+
+public class Configs {
+
+  private Configs() {}
+
+  public static Object put(Map<String, Object> map, Object value, String... path) {
+    String joinedPath = String.join(".", path);
+    return map.put(joinedPath, value);
+  }
+
+  public static String put(Map<String, String> map, String value, String... path) {
+    String joinedPath = String.join(".", path);
+    return map.put(joinedPath, value);
+  }
+
+  public static String putIfAbsent(Map<String, String> map, String value, String... path) {
+    String joinedPath = String.join(".", path);
+    return map.putIfAbsent(joinedPath, value);
+  }
+}

dataSource/filestore/src/main/java/cn/edu/tsinghua/iginx/filestore/common/Fields.java

@@ -21,6 +21,7 @@
 import cn.edu.tsinghua.iginx.engine.shared.data.read.Field;
 import cn.edu.tsinghua.iginx.thrift.DataType;
 import java.util.Map;
+import javax.annotation.Nullable;
 
 public class Fields {
 
@@ -32,4 +33,8 @@ public static Field of(Column column) {
     DataType dataType = column.getDataType();
     return new Field(name, dataType, tags);
   }
+
+  public static Field addPrefix(Field field, @Nullable String prefix) {
+    return new Field(IginxPaths.join(prefix, field.getName()), field.getType(), field.getTags());
+  }
 }

dataSource/filestore/src/main/java/cn/edu/tsinghua/iginx/filestore/common/FileStoreRowStream.java

@@ -0,0 +1,37 @@
+/*
+ * IGinX - the polystore system with high performance
+ * Copyright (C) Tsinghua University
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package cn.edu.tsinghua.iginx.filestore.common;
+
+import cn.edu.tsinghua.iginx.engine.shared.data.read.Header;
+import cn.edu.tsinghua.iginx.engine.shared.data.read.Row;
+import cn.edu.tsinghua.iginx.engine.shared.data.read.RowStream;
+
+public abstract class FileStoreRowStream implements RowStream {
+
+  @Override
+  public abstract Header getHeader() throws FileStoreException;
+
+  @Override
+  public abstract void close() throws FileStoreException;
+
+  @Override
+  public abstract boolean hasNext() throws FileStoreException;
+
+  @Override
+  public abstract Row next() throws FileStoreException;
+}