feat(core): show columns

.github/actions/service/postgresql/action.yml

@@ -63,6 +63,7 @@ runs:
                 echo "port = ${port}" >> "${PGCONF}"
                 echo "unix_socket_directories = ''" >> "${PGCONF}"
                 echo "fsync = off" >> "${PGCONF}"
+                echo "max_connections = 200" >> "${PGCONF}"
 
                 pg_ctl start --pgdata="${PGDATA}" 
               done

core/src/main/java/cn/edu/tsinghua/iginx/engine/physical/storage/IStorage.java

@@ -21,11 +21,17 @@
 import cn.edu.tsinghua.iginx.engine.physical.storage.domain.Column;
 import cn.edu.tsinghua.iginx.engine.physical.storage.domain.DataArea;
 import cn.edu.tsinghua.iginx.engine.physical.task.TaskExecuteResult;
-import cn.edu.tsinghua.iginx.engine.shared.operator.*;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Delete;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Insert;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Project;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Select;
+import cn.edu.tsinghua.iginx.engine.shared.operator.SetTransform;
+import cn.edu.tsinghua.iginx.engine.shared.operator.tag.TagFilter;
 import cn.edu.tsinghua.iginx.metadata.entity.ColumnsInterval;
 import cn.edu.tsinghua.iginx.metadata.entity.KeyInterval;
 import cn.edu.tsinghua.iginx.utils.Pair;
 import java.util.List;
+import java.util.Set;
 
 public interface IStorage {
   /** 对非叠加分片查询数据 */
@@ -60,7 +66,7 @@ default TaskExecuteResult executeProjectWithSetTransform(
   TaskExecuteResult executeInsert(Insert insert, DataArea dataArea);
 
   /** 获取所有列信息 */
-  List<Column> getColumns() throws PhysicalException;
+  List<Column> getColumns(Set<String> patterns, TagFilter tagFilter) throws PhysicalException;
 
   /** 获取指定前缀的数据边界 */
   Pair<ColumnsInterval, KeyInterval> getBoundaryOfStorage(String prefix) throws PhysicalException;

core/src/main/java/cn/edu/tsinghua/iginx/engine/physical/storage/execute/StoragePhysicalTaskExecutor.java

@@ -32,13 +32,18 @@
 import cn.edu.tsinghua.iginx.engine.physical.storage.domain.Column;
 import cn.edu.tsinghua.iginx.engine.physical.storage.domain.DataArea;
 import cn.edu.tsinghua.iginx.engine.physical.storage.queue.StoragePhysicalTaskQueue;
-import cn.edu.tsinghua.iginx.engine.physical.storage.utils.TagKVUtils;
 import cn.edu.tsinghua.iginx.engine.physical.task.GlobalPhysicalTask;
 import cn.edu.tsinghua.iginx.engine.physical.task.MemoryPhysicalTask;
 import cn.edu.tsinghua.iginx.engine.physical.task.StoragePhysicalTask;
 import cn.edu.tsinghua.iginx.engine.physical.task.TaskExecuteResult;
 import cn.edu.tsinghua.iginx.engine.shared.data.read.RowStream;
-import cn.edu.tsinghua.iginx.engine.shared.operator.*;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Delete;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Insert;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Operator;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Project;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Select;
+import cn.edu.tsinghua.iginx.engine.shared.operator.SetTransform;
+import cn.edu.tsinghua.iginx.engine.shared.operator.ShowColumns;
 import cn.edu.tsinghua.iginx.engine.shared.operator.tag.TagFilter;
 import cn.edu.tsinghua.iginx.engine.shared.operator.type.OperatorType;
 import cn.edu.tsinghua.iginx.metadata.DefaultMetaManager;
@@ -353,64 +358,76 @@ public TaskExecuteResult executeGlobalTask(GlobalPhysicalTask task) {
 
   public TaskExecuteResult executeShowColumns(ShowColumns showColumns) {
     List<StorageEngineMeta> storageList = metaManager.getStorageEngineList();
-    Set<Column> columnSet = new HashSet<>();
+    TreeSet<Column> columnSetAfterFilter =
+        new TreeSet<>(Comparator.comparing(Column::getPhysicalPath));
     for (StorageEngineMeta storage : storageList) {
       long id = storage.getId();
       Pair<IStorage, ThreadPoolExecutor> pair = storageManager.getStorage(id);
       if (pair == null) {
         continue;
       }
       try {
-        List<Column> columnList = pair.k.getColumns();
-        // fix the schemaPrefix
+        Set<String> patterns = showColumns.getPathRegexSet();
+        TagFilter tagFilter = showColumns.getTagFilter();
+
         String schemaPrefix = storage.getSchemaPrefix();
-        if (schemaPrefix != null) {
-          for (Column column : columnList) {
-            if (column.isDummy()) {
-              column.setPath(schemaPrefix + "." + column.getPath());
+        // 不下推dataPrefix的原因：非dummy的数据库中如果有原始数据，会在show columns时被查询到并造成误解
+        String dataPrefixRegex =
+            storage.getDataPrefix() == null
+                ? null
+                : StringUtils.reformatPath(storage.getDataPrefix() + ".*");
+
+        // schemaPrefix是在IGinX中定义的，数据源的路径中没有该前缀，因此需要剪掉前缀是schemaPrefix的部分
+        Set<String> patternsCutSchemaPrefix = StringUtils.cutSchemaPrefix(schemaPrefix, patterns);
+        if (patternsCutSchemaPrefix.isEmpty()) {
+          continue;
+        }
+        if (patternsCutSchemaPrefix.contains("*")) {
+          patternsCutSchemaPrefix = Collections.emptySet();
+        }
+        List<Column> columnList = pair.k.getColumns(patternsCutSchemaPrefix, tagFilter);
+
+        if (tagFilter != null) {
+          columnSetAfterFilter.addAll(columnList);
+          continue;
+        }
+        for (Column column : columnList) {
+          if (!column.isDummy()) {
+            columnSetAfterFilter.add(column);
+            continue;
+          }
+          if (dataPrefixRegex == null || Pattern.matches(dataPrefixRegex, column.getPath())) {
+            if (schemaPrefix == null) {
+              columnSetAfterFilter.add(column);
+              continue;
+            }
+            column.setPath(schemaPrefix + "." + column.getPath());
+            boolean isMatch = patterns.isEmpty();
+            for (String pathRegex : patterns) {
+              if (Pattern.matches(StringUtils.reformatPath(pathRegex), column.getPath())) {
+                isMatch = true;
+                break;
+              }
+            }
+            if (isMatch) {
+              columnSetAfterFilter.add(column);
             }
           }
         }
-        columnSet.addAll(columnList);
       } catch (PhysicalException e) {
         return new TaskExecuteResult(e);
       }
     }
 
-    Set<String> pathRegexSet = showColumns.getPathRegexSet();
-    TagFilter tagFilter = showColumns.getTagFilter();
-
-    TreeSet<Column> tsSetAfterFilter = new TreeSet<>(Comparator.comparing(Column::getPhysicalPath));
-    for (Column column : columnSet) {
-      boolean isTarget = true;
-      if (!pathRegexSet.isEmpty()) {
-        isTarget = false;
-        for (String pathRegex : pathRegexSet) {
-          if (Pattern.matches(StringUtils.reformatPath(pathRegex), column.getPath())) {
-            isTarget = true;
-            break;
-          }
-        }
-      }
-      if (tagFilter != null) {
-        if (!TagKVUtils.match(column.getTags(), tagFilter)) {
-          isTarget = false;
-        }
-      }
-      if (isTarget) {
-        tsSetAfterFilter.add(column);
-      }
-    }
-
     int limit = showColumns.getLimit();
     int offset = showColumns.getOffset();
     if (limit == Integer.MAX_VALUE && offset == 0) {
-      return new TaskExecuteResult(Column.toRowStream(tsSetAfterFilter));
+      return new TaskExecuteResult(Column.toRowStream(columnSetAfterFilter));
     } else {
       // only need part of data.
       List<Column> tsList = new ArrayList<>();
-      int cur = 0, size = tsSetAfterFilter.size();
-      for (Iterator<Column> iter = tsSetAfterFilter.iterator(); iter.hasNext(); cur++) {
+      int cur = 0, size = columnSetAfterFilter.size();
+      for (Iterator<Column> iter = columnSetAfterFilter.iterator(); iter.hasNext(); cur++) {
         if (cur >= size || cur - offset >= limit) {
           break;
         }

dataSource/filesystem/src/main/java/cn/edu/tsinghua/iginx/filesystem/FileSystemStorage.java

@@ -26,11 +26,15 @@
 import cn.edu.tsinghua.iginx.engine.physical.storage.domain.Column;
 import cn.edu.tsinghua.iginx.engine.physical.storage.domain.DataArea;
 import cn.edu.tsinghua.iginx.engine.physical.task.TaskExecuteResult;
-import cn.edu.tsinghua.iginx.engine.shared.operator.*;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Delete;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Insert;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Project;
+import cn.edu.tsinghua.iginx.engine.shared.operator.Select;
 import cn.edu.tsinghua.iginx.engine.shared.operator.filter.AndFilter;
 import cn.edu.tsinghua.iginx.engine.shared.operator.filter.Filter;
 import cn.edu.tsinghua.iginx.engine.shared.operator.filter.KeyFilter;
 import cn.edu.tsinghua.iginx.engine.shared.operator.filter.Op;
+import cn.edu.tsinghua.iginx.engine.shared.operator.tag.TagFilter;
 import cn.edu.tsinghua.iginx.filesystem.exec.Executor;
 import cn.edu.tsinghua.iginx.filesystem.exec.LocalExecutor;
 import cn.edu.tsinghua.iginx.filesystem.exec.RemoteExecutor;
@@ -42,6 +46,7 @@
 import cn.edu.tsinghua.iginx.utils.Pair;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Set;
 import org.apache.thrift.transport.TTransportException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -148,8 +153,9 @@ public TaskExecuteResult executeDelete(Delete delete, DataArea dataArea) {
   }
 
   @Override
-  public List<Column> getColumns() throws PhysicalException {
-    return executor.getColumnsOfStorageUnit(WILDCARD);
+  public List<Column> getColumns(Set<String> patterns, TagFilter tagFilter)
+      throws PhysicalException {
+    return executor.getColumnsOfStorageUnit(WILDCARD, patterns, tagFilter);
   }
 
   @Override
@@ -159,7 +165,7 @@ public Pair<ColumnsInterval, KeyInterval> getBoundaryOfStorage(String prefix)
   }
 
   @Override
-  public synchronized void release() throws PhysicalException {
+  public synchronized void release() {
     executor.close();
     if (thread != null) {
       thread.interrupt();

dataSource/filesystem/src/main/java/cn/edu/tsinghua/iginx/filesystem/exec/Executor.java

@@ -29,6 +29,7 @@
 import cn.edu.tsinghua.iginx.metadata.entity.KeyInterval;
 import cn.edu.tsinghua.iginx.utils.Pair;
 import java.util.List;
+import java.util.Set;
 
 public interface Executor {
 
@@ -44,7 +45,8 @@ TaskExecuteResult executeProjectTask(
   TaskExecuteResult executeDeleteTask(
       List<String> paths, List<KeyRange> keyRanges, TagFilter tagFilter, String storageUnit);
 
-  List<Column> getColumnsOfStorageUnit(String storageUnit) throws PhysicalException;
+  List<Column> getColumnsOfStorageUnit(
+      String storageUnit, Set<String> patterns, TagFilter tagFilter) throws PhysicalException;
 
   Pair<ColumnsInterval, KeyInterval> getBoundaryOfStorage(String dataPrefix)
       throws PhysicalException;

dataSource/filesystem/src/main/java/cn/edu/tsinghua/iginx/filesystem/exec/LocalExecutor.java

@@ -25,6 +25,7 @@
 import cn.edu.tsinghua.iginx.engine.physical.exception.PhysicalException;
 import cn.edu.tsinghua.iginx.engine.physical.memory.execute.stream.EmptyRowStream;
 import cn.edu.tsinghua.iginx.engine.physical.storage.domain.Column;
+import cn.edu.tsinghua.iginx.engine.physical.storage.utils.TagKVUtils;
 import cn.edu.tsinghua.iginx.engine.physical.task.TaskExecuteResult;
 import cn.edu.tsinghua.iginx.engine.shared.KeyRange;
 import cn.edu.tsinghua.iginx.engine.shared.data.read.RowStream;
@@ -47,11 +48,13 @@
 import cn.edu.tsinghua.iginx.metadata.entity.KeyInterval;
 import cn.edu.tsinghua.iginx.thrift.DataType;
 import cn.edu.tsinghua.iginx.utils.Pair;
+import cn.edu.tsinghua.iginx.utils.StringUtils;
 import java.io.File;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -67,9 +70,9 @@ public class LocalExecutor implements Executor {
 
   private String prefix;
 
-  private boolean hasData;
+  private final boolean hasData;
 
-  private FileSystemManager fileSystemManager;
+  private final FileSystemManager fileSystemManager;
 
   public LocalExecutor(boolean isReadOnly, boolean hasData, Map<String, String> extraParams) {
     String dir = extraParams.get(Constant.INIT_INFO_DIR);
@@ -327,43 +330,48 @@ public TaskExecuteResult executeDeleteTask(
   }
 
   @Override
-  public List<Column> getColumnsOfStorageUnit(String storageUnit) throws PhysicalException {
+  public List<Column> getColumnsOfStorageUnit(
+      String storageUnit, Set<String> patterns, TagFilter tagFilter) throws PhysicalException {
     List<Column> columns = new ArrayList<>();
     if (root != null) {
       File directory = new File(FilePathUtils.toIginxPath(root, storageUnit, null));
       for (File file : fileSystemManager.getAllFiles(directory, false)) {
         FileMeta meta = fileSystemManager.getFileMeta(file);
+        String columnPath =
+            FilePathUtils.convertAbsolutePathToPath(root, file.getAbsolutePath(), storageUnit);
         if (meta == null) {
           throw new PhysicalException(
               String.format(
                   "encounter error when getting columns of storage unit because file meta %s is null",
                   file.getAbsolutePath()));
         }
-        columns.add(
-            new Column(
-                FilePathUtils.convertAbsolutePathToPath(root, file.getAbsolutePath(), storageUnit),
-                meta.getDataType(),
-                meta.getTags(),
-                false));
+        // get columns by pattern
+        if (StringUtils.pathNotMatchPatterns(columnPath, patterns)) {
+          continue;
+        }
+        // get columns by tag filter
+        if (tagFilter != null && !TagKVUtils.match(meta.getTags(), tagFilter)) {
+          continue;
+        }
+        columns.add(new Column(columnPath, meta.getDataType(), meta.getTags(), false));
       }
     }
-    if (hasData && dummyRoot != null) {
+    // get columns from dummy storage unit
+    if (hasData && dummyRoot != null && tagFilter == null) {
       for (File file : fileSystemManager.getAllFiles(new File(realDummyRoot), true)) {
-        columns.add(
-            new Column(
-                FilePathUtils.convertAbsolutePathToPath(
-                    dummyRoot, file.getAbsolutePath(), storageUnit),
-                DataType.BINARY,
-                null,
-                true));
+        String dummyPath =
+            FilePathUtils.convertAbsolutePathToPath(dummyRoot, file.getAbsolutePath(), storageUnit);
+        if (StringUtils.pathNotMatchPatterns(dummyPath, patterns)) {
+          continue;
+        }
+        columns.add(new Column(dummyPath, DataType.BINARY, null, true));
       }
     }
     return columns;
   }
 
   @Override
-  public Pair<ColumnsInterval, KeyInterval> getBoundaryOfStorage(String dataPrefix)
-      throws PhysicalException {
+  public Pair<ColumnsInterval, KeyInterval> getBoundaryOfStorage(String dataPrefix) {
     KeyInterval keyInterval = KeyInterval.getDefaultKeyInterval();
     ColumnsInterval columnsInterval;
 

dataSource/filesystem/src/main/java/cn/edu/tsinghua/iginx/filesystem/exec/RemoteExecutor.java

@@ -31,10 +31,25 @@
 import cn.edu.tsinghua.iginx.engine.shared.data.write.DataView;
 import cn.edu.tsinghua.iginx.engine.shared.data.write.RawDataType;
 import cn.edu.tsinghua.iginx.engine.shared.operator.filter.Filter;
-import cn.edu.tsinghua.iginx.engine.shared.operator.tag.*;
+import cn.edu.tsinghua.iginx.engine.shared.operator.tag.AndTagFilter;
+import cn.edu.tsinghua.iginx.engine.shared.operator.tag.BasePreciseTagFilter;
+import cn.edu.tsinghua.iginx.engine.shared.operator.tag.BaseTagFilter;
+import cn.edu.tsinghua.iginx.engine.shared.operator.tag.OrTagFilter;
+import cn.edu.tsinghua.iginx.engine.shared.operator.tag.PreciseTagFilter;
+import cn.edu.tsinghua.iginx.engine.shared.operator.tag.TagFilter;
 import cn.edu.tsinghua.iginx.filesystem.exception.FilesystemException;
-import cn.edu.tsinghua.iginx.filesystem.thrift.*;
+import cn.edu.tsinghua.iginx.filesystem.thrift.DeleteReq;
+import cn.edu.tsinghua.iginx.filesystem.thrift.FSHeader;
+import cn.edu.tsinghua.iginx.filesystem.thrift.FSKeyRange;
+import cn.edu.tsinghua.iginx.filesystem.thrift.FSRawData;
 import cn.edu.tsinghua.iginx.filesystem.thrift.FileSystemService.Client;
+import cn.edu.tsinghua.iginx.filesystem.thrift.GetBoundaryOfStorageResp;
+import cn.edu.tsinghua.iginx.filesystem.thrift.GetColumnsOfStorageUnitResp;
+import cn.edu.tsinghua.iginx.filesystem.thrift.InsertReq;
+import cn.edu.tsinghua.iginx.filesystem.thrift.ProjectReq;
+import cn.edu.tsinghua.iginx.filesystem.thrift.ProjectResp;
+import cn.edu.tsinghua.iginx.filesystem.thrift.RawTagFilter;
+import cn.edu.tsinghua.iginx.filesystem.thrift.Status;
 import cn.edu.tsinghua.iginx.filesystem.thrift.TagFilterType;
 import cn.edu.tsinghua.iginx.filesystem.tools.FilterTransformer;
 import cn.edu.tsinghua.iginx.metadata.entity.ColumnsInterval;
@@ -50,6 +65,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.thrift.TException;
 import org.apache.thrift.protocol.TBinaryProtocol;
@@ -225,11 +241,13 @@ public TaskExecuteResult executeDeleteTask(
   }
 
   @Override
-  public List<Column> getColumnsOfStorageUnit(String storageUnit) throws PhysicalException {
+  public List<Column> getColumnsOfStorageUnit(
+      String storageUnit, Set<String> patterns, TagFilter tagFilter) throws PhysicalException {
     try {
       TTransport transport = thriftConnPool.borrowTransport();
       Client client = new Client(new TBinaryProtocol(transport));
-      GetColumnsOfStorageUnitResp resp = client.getColumnsOfStorageUnit(storageUnit);
+      GetColumnsOfStorageUnitResp resp =
+          client.getColumnsOfStorageUnit(storageUnit, patterns, constructRawTagFilter(tagFilter));
       thriftConnPool.returnTransport(transport);
       List<Column> columns = new ArrayList<>();
       resp.getPathList()
@@ -272,6 +290,9 @@ public void close() {
 
   private RawTagFilter constructRawTagFilter(TagFilter tagFilter) {
     RawTagFilter filter = null;
+    if (tagFilter == null) {
+      return null;
+    }
     switch (tagFilter.getType()) {
       case Base:
         {