Repository: diennea/herddb (java)

Issue:
Support jdbc index metadata
Support method DatabaseMetaData.getIndexInfo

Pull request description:
ISSUE-188 Implement JDBC getIndexInfo
Add sysindexcolumns systemtable
Enforce sort by PK of system tables (in order to support joins, as Calcite assumes that records are sorted)
Implement JDBC getIndexInfo

closes #188

Patch:
diff --git a/herddb-core/src/main/java/herddb/core/TableSpaceManager.java b/herddb-core/src/main/java/herddb/core/TableSpaceManager.java
index 87bfa13f..ac892015 100644
--- a/herddb-core/src/main/java/herddb/core/TableSpaceManager.java
+++ b/herddb-core/src/main/java/herddb/core/TableSpaceManager.java
@@ -57,6 +57,7 @@ import herddb.core.stats.TableSpaceManagerStats;
 import herddb.core.system.SysclientsTableManager;
 import herddb.core.system.SyscolumnsTableManager;
 import herddb.core.system.SysconfigTableManager;
+import herddb.core.system.SysindexcolumnsTableManager;
 import herddb.core.system.SysindexesTableManager;
 import herddb.core.system.SysnodesTableManager;
 import herddb.core.system.SystablesTableManager;
@@ -185,6 +186,7 @@ public class TableSpaceManager {
             registerSystemTableManager(new SystablesTableManager(this));
             registerSystemTableManager(new SystablestatsTableManager(this));
             registerSystemTableManager(new SysindexesTableManager(this));
+            registerSystemTableManager(new SysindexcolumnsTableManager(this));
             registerSystemTableManager(new SyscolumnsTableManager(this));
             registerSystemTableManager(new SystransactionsTableManager(this));
         }
diff --git a/herddb-core/src/main/java/herddb/core/system/AbstractSystemTableManager.java b/herddb-core/src/main/java/herddb/core/system/AbstractSystemTableManager.java
index 2148844d..89b3f38d 100644
--- a/herddb-core/src/main/java/herddb/core/system/AbstractSystemTableManager.java
+++ b/herddb-core/src/main/java/herddb/core/system/AbstractSystemTableManager.java
@@ -47,6 +47,10 @@ import herddb.model.Transaction;
 import herddb.model.commands.ScanStatement;
 import herddb.storage.DataStorageManagerException;
 import herddb.storage.FullTableScanConsumer;
+import herddb.utils.SQLRecordPredicateFunctions;
+import java.util.Comparator;
+import java.util.Map;
+import java.util.stream.StreamSupport;
 
 /**
  * System tables
@@ -210,17 +214,41 @@ public abstract class AbstractSystemTableManager implements AbstractTableManager
 
     protected abstract Iterable<Record> buildVirtualRecordList() throws StatementExecutionException;
 
+    private final Comparator<Record> SORT_BY_PK = new Comparator<Record>() {
+        @Override
+        public int compare(Record o1, Record o2) {
+            Map<String, Object> ac1 = o1.toBean(table);
+            Map<String, Object> ac2 = o2.toBean(table);
+
+            for (String pk : table.primaryKey) {
+                Object value1 = ac1.get(pk);
+                Object value2 = ac2.get(pk);
+                int cmp = SQLRecordPredicateFunctions.compare(value1, value2);
+                if (cmp != 0) {
+                    return cmp;
+                }
+            }
+            return 0;
+        }
+    };
+
     @Override
+
     public DataScanner scan(ScanStatement statement, StatementEvaluationContext context,
             Transaction transaction, boolean lockRequired, boolean forWrite) throws StatementExecutionException {
         Predicate predicate = statement.getPredicate();
         MaterializedRecordSet recordSet = tableSpaceManager.getDbmanager().getRecordSetFactory()
-            .createRecordSet(table.columnNames, table.columns);
-        for (Record record : buildVirtualRecordList()) {
-            if (predicate == null || predicate.evaluate(record, context)) {
-                recordSet.add(record.getDataAccessor(table));
-            }
-        }
+                .createRecordSet(table.columnNames, table.columns);
+        Iterable<Record> data = buildVirtualRecordList();
+        StreamSupport
+                .stream(data.spliterator(), false)
+                .filter(record -> {
+                    return (predicate == null || predicate.evaluate(record, context));
+                })
+                .sorted(SORT_BY_PK) // enforce sort by PK
+                .map(r -> r.getDataAccessor(table))
+                .forEach(recordSet::add);
+
         recordSet.writeFinished();
         recordSet.sort(statement.getComparator());
         recordSet.applyLimits(statement.getLimits(), context);
diff --git a/herddb-core/src/main/java/herddb/core/system/SysindexcolumnsTableManager.java b/herddb-core/src/main/java/herddb/core/system/SysindexcolumnsTableManager.java
new file mode 100644
index 00000000..98bfb140
--- /dev/null
+++ b/herddb-core/src/main/java/herddb/core/system/SysindexcolumnsTableManager.java
@@ -0,0 +1,112 @@
+/*
+ Licensed to Diennea S.r.l. under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. Diennea S.r.l. licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License.  You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied.  See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+ */
+package herddb.core.system;
+
+import herddb.codec.RecordSerializer;
+import herddb.core.AbstractIndexManager;
+import herddb.core.TableSpaceManager;
+import herddb.index.blink.BLinkKeyToPageIndex;
+import herddb.model.Column;
+import herddb.model.ColumnTypes;
+import herddb.model.Index;
+import herddb.model.Record;
+import herddb.model.Table;
+import herddb.utils.SQLRecordPredicateFunctions;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Table Manager for the SYSINDEXCOLUMNS virtual table
+ *
+ * @author enrico.olivelli
+ */
+public class SysindexcolumnsTableManager extends AbstractSystemTableManager {
+
+    private final static Table TABLE = Table
+            .builder()
+            .name("sysindexcolumns")
+            .column("tablespace", ColumnTypes.STRING)
+            .column("table_name", ColumnTypes.STRING)
+            .column("index_name", ColumnTypes.STRING)
+            .column("index_uuid", ColumnTypes.STRING)
+            .column("index_type", ColumnTypes.STRING)
+            .column("column_name", ColumnTypes.STRING)
+            .column("ordinal_position", ColumnTypes.INTEGER)
+            .column("clustered", ColumnTypes.INTEGER)
+            .column("unique", ColumnTypes.INTEGER)
+            .primaryKey("table_name", false)
+            .primaryKey("index_name", false)
+            .primaryKey("column_name", false)
+            .build();
+
+    public SysindexcolumnsTableManager(TableSpaceManager parent) {
+        super(parent, TABLE);
+    }
+
+    @Override
+    protected Iterable<Record> buildVirtualRecordList() {
+        List<Table> tables = tableSpaceManager.getAllCommittedTables();
+        List<Record> result = new ArrayList<>();
+        tables.forEach(r -> {
+            // PK
+            int posPk = 0;
+            for (String pk : r.getPrimaryKey()) {
+                Column column = r.getColumn(pk);
+                String indexName = BLinkKeyToPageIndex.deriveIndexName(r.name);
+                result.add(RecordSerializer.makeRecord(table,
+                        "index_type", "pk",
+                        "column_name", column.name,
+                        "ordinal_position", posPk++,
+                        "clustered", 1,
+                        "unique", 1,
+                        "tablespace", r.tablespace,
+                        "table_name", r.name,
+                        "index_name", indexName,
+                        "index_uuid", indexName // uuid = index_name in BLink !
+                ));
+            }
+
+            Map<String, AbstractIndexManager> indexesOnTable = tableSpaceManager.getIndexesOnTable(r.name);
+            if (indexesOnTable != null) {
+                indexesOnTable.values().forEach(indexManager -> {
+                    Index index = indexManager.getIndex();
+                    int pos = 0;
+                    for (Column cc : index.getColumns()) {
+                        result.add(RecordSerializer.makeRecord(table,
+                                "tablespace", r.tablespace,
+                                "table_name", r.name,
+                                "index_name", index.name,
+                                "index_uuid", index.uuid,
+                                "index_type", index.type,
+                                "column_name", cc.name,
+                                "ordinal_position", pos++,
+                                "clustered", 0,
+                                "unique", 0
+                        ));
+                    }
+                });
+            }
+        });
+        return result;
+    }
+
+}
diff --git a/herddb-core/src/main/java/herddb/core/system/SystablesTableManager.java b/herddb-core/src/main/java/herddb/core/system/SystablesTableManager.java
index 97663f6a..54c2c546 100644
--- a/herddb-core/src/main/java/herddb/core/system/SystablesTableManager.java
+++ b/herddb-core/src/main/java/herddb/core/system/SystablesTableManager.java
@@ -35,15 +35,15 @@ import java.util.stream.Collectors;
 public class SystablesTableManager extends AbstractSystemTableManager {
 
     private final static Table TABLE = Table
-        .builder()
-        .name("systables")
-        .column("tablespace", ColumnTypes.STRING)
-        .column("table_name", ColumnTypes.STRING)
-        .column("table_uuid", ColumnTypes.STRING)
-        .column("systemtable", ColumnTypes.STRING)
-        .primaryKey("tablespace", false)
-        .primaryKey("table_name", false)
-        .build();
+            .builder()
+            .name("systables")
+            .column("tablespace", ColumnTypes.STRING)
+            .column("table_name", ColumnTypes.STRING)
+            .column("table_uuid", ColumnTypes.STRING)
+            .column("systemtable", ColumnTypes.STRING)
+            .primaryKey("tablespace", false)
+            .primaryKey("table_name", false)
+            .build();
 
     public SystablesTableManager(TableSpaceManager parent) {
         super(parent, TABLE);
@@ -53,15 +53,15 @@ public class SystablesTableManager extends AbstractSystemTableManager {
     protected Iterable<Record> buildVirtualRecordList() {
         List<Table> tables = tableSpaceManager.getAllCommittedTables();
         return tables
-            .stream()
-            .map(r -> RecordSerializer.makeRecord(table,
-            "tablespace", r.tablespace,
-            "table_name", r.name,
-            "table_uuid", r.uuid,
-            "systemtable",
-            r.name.startsWith("sys") ? "true" : "false"
+                .stream()
+                .map(r -> RecordSerializer.makeRecord(table,
+                "tablespace", r.tablespace,
+                "table_name", r.name,
+                "table_uuid", r.uuid,
+                "systemtable",
+                r.name.startsWith("sys") ? "true" : "false"
         ))
-            .collect(Collectors.toList());
+                .collect(Collectors.toList());
     }
 
 }
diff --git a/herddb-core/src/main/java/herddb/index/blink/BLinkKeyToPageIndex.java b/herddb-core/src/main/java/herddb/index/blink/BLinkKeyToPageIndex.java
index eb10ee25..ba2a4f6a 100644
--- a/herddb-core/src/main/java/herddb/index/blink/BLinkKeyToPageIndex.java
+++ b/herddb-core/src/main/java/herddb/index/blink/BLinkKeyToPageIndex.java
@@ -97,10 +97,14 @@ public class BLinkKeyToPageIndex implements KeyToPageIndex {
 
     private final AtomicBoolean closed;
 
+    public static String deriveIndexName(String tableName) {
+        return tableName + "_primary";
+    }
+    
     public BLinkKeyToPageIndex(String tableSpace, String tableName, MemoryManager memoryManager, DataStorageManager dataStorageManager) {
         super();
         this.tableSpace = tableSpace;
-        this.indexName = tableName + "_primary";
+        this.indexName = deriveIndexName(tableName);
 
         this.memoryManager = memoryManager;
         this.dataStorageManager = dataStorageManager;
diff --git a/herddb-jdbc/src/main/java/herddb/jdbc/HerdDBDatabaseMetadata.java b/herddb-jdbc/src/main/java/herddb/jdbc/HerdDBDatabaseMetadata.java
index f120ec44..2607c660 100644
--- a/herddb-jdbc/src/main/java/herddb/jdbc/HerdDBDatabaseMetadata.java
+++ b/herddb-jdbc/src/main/java/herddb/jdbc/HerdDBDatabaseMetadata.java
@@ -24,7 +24,6 @@ import herddb.client.ScanResultSetMetadata;
 import herddb.client.impl.EmptyScanResultSet;
 import herddb.client.impl.IteratorScanResultSet;
 import herddb.model.TransactionContext;
-import herddb.sql.SQLRecordPredicate;
 import herddb.utils.SQLUtils;
 import java.sql.Connection;
 import java.sql.DatabaseMetaData;
@@ -33,7 +32,6 @@ import java.sql.RowIdLifetime;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -670,6 +668,20 @@ public class HerdDBDatabaseMetadata implements DatabaseMetaData {
         "SELF_REFERENCING_COL_NAME",
         "REF_GENERATION"
     };
+    private static final String[] GET_INDEXES_SCHEMA = new String[]{
+        "TABLE_CAT",
+        "TABLE_SCHEM",
+        "TABLE_NAME",
+        "NON_UNIQUE",
+        "INDEX_QUALIFIER",
+        "INDEX_NAME",
+        "TYPE",
+        "ORDINAL_POSITION",
+        "COLUMN_NAME",
+        "ASC_OR_DESC",
+        "CARDINALITY",
+        "FILTER_CONDITION"
+    };
 
     @Override
     /**
@@ -1032,9 +1044,113 @@ public class HerdDBDatabaseMetadata implements DatabaseMetaData {
         return new HerdDBResultSet(new EmptyScanResultSet(TransactionContext.NOTRANSACTION_ID));
     }
 
+    /**
+     * Retrieves a description of the given table's indices and statistics. They
+     * are ordered by NON_UNIQUE, TYPE, INDEX_NAME, and ORDINAL_POSITION.
+     *
+     * <P>
+     * Each index column description has the following columns:
+     * <OL>
+     * <LI><B>TABLE_CAT</B> String {@code =>} table catalog (may be
+     * <code>null</code>)
+     * <LI><B>TABLE_SCHEM</B> String {@code =>} table schema (may be
+     * <code>null</code>)
+     * <LI><B>TABLE_NAME</B> String {@code =>} table name
+     * <LI><B>NON_UNIQUE</B> boolean {@code =>} Can index values be non-unique.
+     * false when TYPE is tableIndexStatistic
+     * <LI><B>INDEX_QUALIFIER</B> String {@code =>} index catalog (may be
+     * <code>null</code>); <code>null</code> when TYPE is tableIndexStatistic
+     * <LI><B>INDEX_NAME</B> String {@code =>} index name; <code>null</code>
+     * when TYPE is tableIndexStatistic
+     * <LI><B>TYPE</B> short {@code =>} index type:
+     * <UL>
+     * <LI> tableIndexStatistic - this identifies table statistics that are
+     * returned in conjunction with a table's index descriptions
+     * <LI> tableIndexClustered - this is a clustered index
+     * <LI> tableIndexHashed - this is a hashed index
+     * <LI> tableIndexOther - this is some other style of index
+     * </UL>
+     * <LI><B>ORDINAL_POSITION</B> short {@code =>} column sequence number
+     * within index; zero when TYPE is tableIndexStatistic
+     * <LI><B>COLUMN_NAME</B> String {@code =>} column name; <code>null</code>
+     * when TYPE is tableIndexStatistic
+     * <LI><B>ASC_OR_DESC</B> String {@code =>} column sort sequence, "A"
+     * {@code =>} ascending, "D" {@code =>} descending, may be <code>null</code>
+     * if sort sequence is not supported; <code>null</code> when TYPE is
+     * tableIndexStatistic
+     * <LI><B>CARDINALITY</B> long {@code =>} When TYPE is tableIndexStatistic,
+     * then this is the number of rows in the table; otherwise, it is the number
+     * of unique values in the index.
+     * <LI><B>PAGES</B> long {@code =>} When TYPE is tableIndexStatistic then
+     * this is the number of pages used for the table, otherwise it is the
+     * number of pages used for the current index.
+     * <LI><B>FILTER_CONDITION</B> String {@code =>} Filter condition, if any.
+     * (may be <code>null</code>)
+     * </OL>
+     *
+     * @param catalog a catalog name; must match the catalog name as it is
+     * stored in this database; "" retrieves those without a catalog;
+     * <code>null</code> means that the catalog name should not be used to
+     * narrow the search
+     * @param schema a schema name; must match the schema name as it is stored
+     * in this database; "" retrieves those without a schema; <code>null</code>
+     * means that the schema name should not be used to narrow the search
+     * @param table a table name; must match the table name as it is stored in
+     * this database
+     * @param onlyUnique when true, return only indices for unique values; when
+     * false, return indices regardless of whether unique or not
+     * @param approximate when true, result is allowed to reflect approximate or
+     * out of data values; when false, results are requested to be accurate
+     * @return <code>ResultSet</code> - each row is an index column description
+     * @exception SQLException if a database access error occurs
+     */
     @Override
-    public ResultSet getIndexInfo(String catalog, String schema, String table, boolean unique, boolean approximate) throws SQLException {
-        return new HerdDBResultSet(new EmptyScanResultSet(TransactionContext.NOTRANSACTION_ID));
+    public ResultSet getIndexInfo(String catalog, String schema, String tableNamePattern, boolean onlyUnique, boolean approximate) throws SQLException {
+        String query = "SELECT * FROM SYSINDEXCOLUMNS";
+        if (tableNamePattern != null && !tableNamePattern.isEmpty()) {
+            query = query + " WHERE table_name LIKE '" + SQLUtils.escape(tableNamePattern) + "'";
+        }
+        System.out.println("query: " + query);
+        try (Statement statement = con.createStatement();
+                ResultSet rs = statement.executeQuery(query)) {
+
+            List<Map<String, Object>> results = new ArrayList<>();
+            while (rs.next()) {
+                String table_name = rs.getString("table_name");
+                String index_name = rs.getString("index_name");
+                String column_name = rs.getString("column_name");
+                int ordinal_position = rs.getInt("ordinal_position");
+                boolean clustered = rs.getInt("clustered") == 1;
+                boolean uniqueValues = rs.getInt("unique") == 1;
+
+                if (onlyUnique && !uniqueValues) {
+                    System.out.println("DISCARD: " + table_name + "." + index_name + " col " + column_name + " clusterd:" + clustered + " uniq " + uniqueValues);
+                    continue;
+                }
+                System.out.println("FOUND: " + table_name + "." + index_name + " col " + column_name + " clusterd:" + clustered + " uniq " + uniqueValues);
+
+                Map<String, Object> data = new HashMap<>();
+                data.put("TABLE_CAT", null);
+                data.put("TABLE_SCHEM", tableSpace);
+                data.put("TABLE_NAME", table_name);
+                data.put("NON_UNIQUE", !uniqueValues);
+                data.put("INDEX_QUALIFIER", null);
+
+                data.put("INDEX_NAME", index_name);
+                data.put("TYPE", clustered ? DatabaseMetaData.tableIndexClustered : DatabaseMetaData.tableIndexOther);
+
+                data.put("ORDINAL_POSITION", ordinal_position);
+
+                data.put("COLUMN_NAME", column_name);
+                data.put("ASC_OR_DESC", null);
+                data.put("CARDINALITY", null);
+                data.put("FILTER_CONDITION", null);
+                results.add(data);
+
+            }
+            ScanResultSetMetadata metadata = new ScanResultSetMetadata(GET_INDEXES_SCHEMA);
+            return new HerdDBResultSet(new IteratorScanResultSet(TransactionContext.NOTRANSACTION_ID, metadata, results.iterator()));
+        }
     }
 
     @Override


Test patch:
diff --git a/herddb-core/src/test/java/herddb/core/SystemTablesTest.java b/herddb-core/src/test/java/herddb/core/SystemTablesTest.java
index a15375d9..9dd39bda 100644
--- a/herddb-core/src/test/java/herddb/core/SystemTablesTest.java
+++ b/herddb-core/src/test/java/herddb/core/SystemTablesTest.java
@@ -100,7 +100,7 @@ public class SystemTablesTest {
             }
 
             try (DataScanner scan = scan(manager, "SELECT * FROM tblspace1.sysindexes order by index_name",
-                Collections.emptyList());) {
+                    Collections.emptyList());) {
                 List<DataAccessor> records = scan.consume();
                 assertEquals(2, records.size());
                 DataAccessor index1 = records.get(0);
@@ -116,10 +116,174 @@ public class SystemTablesTest {
                 assertEquals(RawString.of("tsql2"), index2.get("table_name"));
             }
 
+            try (DataScanner scan = scan(manager, "SELECT * FROM tblspace1.sysindexcolumns order by index_name, column_name",
+                    Collections.emptyList());) {
+                List<DataAccessor> records = scan.consume();
+                for (DataAccessor da : records) {
+                    System.out.println("rec: " + da.toMap());
+                }
+                assertTrue(records
+                        .stream()
+                        .map(d -> d.toMap())
+                        .filter(d -> {
+                            return d.get("table_name").equals("tsql2")
+                                    && d.get("index_type").equals("brin")
+                                    && d.get("column_name").equals("b1")
+                                    && d.get("ordinal_position").equals(1)
+                                    && d.get("index_uuid") != null
+                                    && d.get("tablespace").equals("tblspace1")
+                                    && d.get("unique").equals(0)
+                                    && d.get("index_name").equals("index1");
+                        })
+                        .findAny()
+                        .isPresent());
+                assertTrue(records
+                        .stream()
+                        .map(d -> d.toMap())
+                        .filter(d -> {
+                            return d.get("table_name").equals("tsql2")
+                                    && d.get("index_type").equals("brin")
+                                    && d.get("column_name").equals("s1")
+                                    && d.get("ordinal_position").equals(0)
+                                    && d.get("index_uuid") != null
+                                    && d.get("tablespace").equals("tblspace1")
+                                    && d.get("unique").equals(0)
+                                    && d.get("index_name").equals("index1");
+                        })
+                        .findAny()
+                        .isPresent());
+                assertTrue(records
+                        .stream()
+                        .map(d -> d.toMap())
+                        .filter(d -> {
+                            return d.get("table_name").equals("tsql2")
+                                    && d.get("index_type").equals("hash")
+                                    && d.get("column_name").equals("b1")
+                                    && d.get("ordinal_position").equals(0)
+                                    && d.get("index_uuid") != null
+                                    && d.get("tablespace").equals("tblspace1")
+                                    && d.get("unique").equals(0)
+                                    && d.get("index_name").equals("index2");
+                        })
+                        .findAny()
+                        .isPresent());
+                assertTrue(records
+                        .stream()
+                        .map(d -> d.toMap())
+                        .filter(d -> {
+                            return d.get("table_name").equals("tsql2")
+                                    && d.get("index_type").equals("pk")
+                                    && d.get("column_name").equals("k1")
+                                    && d.get("ordinal_position").equals(0)
+                                    && d.get("index_uuid").equals("tsql2_primary") // index_uuid == index_name for BLink
+                                    && d.get("tablespace").equals("tblspace1")
+                                    && d.get("unique").equals(1)
+                                    && d.get("index_name").equals("tsql2_primary");
+                        })
+                        .findAny()
+                        .isPresent());
+                assertEquals(21, records.size());
+            }
+            
+            try (DataScanner scan = scan(manager, "SELECT * FROM tblspace1.sysindexcolumns where table_name like '%tsql' order by index_name, column_name",
+                    Collections.emptyList());) {
+                List<DataAccessor> records = scan.consume();
+                for (DataAccessor da : records) {
+                    System.out.println("rec: " + da.toMap());
+                }                
+                assertTrue(records
+                        .stream()
+                        .map(d -> d.toMap())
+                        .filter(d -> {
+                            return d.get("table_name").equals("tsql")
+                                    && d.get("index_type").equals("pk")
+                                    && d.get("column_name").equals("k1")
+                                    && d.get("ordinal_position").equals(0)
+                                    && d.get("index_uuid").equals("tsql_primary") // index_uuid == index_name for BLink
+                                    && d.get("tablespace").equals("tblspace1")
+                                    && d.get("unique").equals(1)
+                                    && d.get("index_name").equals("tsql_primary");
+                        })
+                        .findAny()
+                        .isPresent());
+                assertEquals(1, records.size());
+            }
+
+            try (DataScanner scan = scan(manager, "SELECT sc.* FROM tblspace1.sysindexcolumns sc "
+                    + "JOIN tblspace1.systables st on sc.table_name=st.table_name and st.systemtable = 'false' " // only non system tables
+                    + "order by sc.index_name, sc.column_name",
+                    Collections.emptyList());) {
+                List<DataAccessor> records = scan.consume();
+                for (DataAccessor da : records) {
+                    System.out.println("rec: " + da.toMap());
+                }
+                assertTrue(records
+                        .stream()
+                        .map(d -> d.toMap())
+                        .filter(d -> {
+                            return d.get("table_name").equals("tsql2")
+                                    && d.get("index_type").equals("brin")
+                                    && d.get("column_name").equals("b1")
+                                    && d.get("ordinal_position").equals(1)
+                                    && d.get("index_uuid") != null
+                                    && d.get("tablespace").equals("tblspace1")
+                                    && d.get("unique").equals(0)
+                                    && d.get("index_name").equals("index1");
+                        })
+                        .findAny()
+                        .isPresent());
+                assertTrue(records
+                        .stream()
+                        .map(d -> d.toMap())
+                        .filter(d -> {
+                            return d.get("table_name").equals("tsql2")
+                                    && d.get("index_type").equals("brin")
+                                    && d.get("column_name").equals("s1")
+                                    && d.get("ordinal_position").equals(0)
+                                    && d.get("index_uuid") != null
+                                    && d.get("tablespace").equals("tblspace1")
+                                    && d.get("unique").equals(0)
+                                    && d.get("index_name").equals("index1");
+                        })
+                        .findAny()
+                        .isPresent());
+                assertTrue(records
+                        .stream()
+                        .map(d -> d.toMap())
+                        .filter(d -> {
+                            return d.get("table_name").equals("tsql2")
+                                    && d.get("index_type").equals("hash")
+                                    && d.get("column_name").equals("b1")
+                                    && d.get("ordinal_position").equals(0)
+                                    && d.get("index_uuid") != null
+                                    && d.get("tablespace").equals("tblspace1")
+                                    && d.get("unique").equals(0)
+                                    && d.get("index_name").equals("index2");
+                        })
+                        .findAny()
+                        .isPresent());
+                assertTrue(records
+                        .stream()
+                        .map(d -> d.toMap())
+                        .filter(d -> {
+                            return d.get("table_name").equals("tsql2")
+                                    && d.get("index_type").equals("pk")
+                                    && d.get("column_name").equals("k1")
+                                    && d.get("ordinal_position").equals(0)
+                                    && d.get("index_uuid").equals("tsql2_primary") // index_uuid == index_name for BLink
+                                    && d.get("tablespace").equals("tblspace1")
+                                    && d.get("unique").equals(1)
+                                    && d.get("index_name").equals("tsql2_primary");
+                        })
+                        .findAny()
+                        .isPresent());
+                assertEquals(5, records.size());
+            }
+
             execute(manager, "BEGIN TRANSACTION 'tblspace1'", Collections.emptyList());
             long txid;
             try (DataScanner scan = scan(manager, "SELECT * FROM tblspace1.systransactions order by txid",
-                Collections.emptyList());) {
+                    Collections.emptyList());) {
                 List<DataAccessor> records = scan.consume();
                 assertEquals(1, records.size());
                 System.out.println("records:" + records);
@@ -127,7 +291,7 @@ public class SystemTablesTest {
             }
             execute(manager, "COMMIT TRANSACTION 'tblspace1'," + txid, Collections.emptyList());
             try (DataScanner scan = scan(manager, "SELECT * FROM tblspace1.systransactions order by txid",
-                Collections.emptyList());) {
+                    Collections.emptyList());) {
                 List<DataAccessor> records = scan.consume();
                 assertEquals(0, records.size());
             }
@@ -138,37 +302,37 @@ public class SystemTablesTest {
                     System.out.println("found " + r.toMap());
                 });
                 assertTrue(records.stream()
-                    .filter(t
-                        -> t.get("table_name").equals("tsql")
-                    && t.get("column_name").equals("k1")
-                    && t.get("data_type").equals("string")
-                    && t.get("auto_increment").equals(1)
-                    ).findAny().isPresent());
+                        .filter(t
+                                -> t.get("table_name").equals("tsql")
+                        && t.get("column_name").equals("k1")
+                        && t.get("data_type").equals("string")
+                        && t.get("auto_increment").equals(1)
+                        ).findAny().isPresent());
                 assertTrue(records.stream()
-                    .filter(t
-                        -> t.get("table_name").equals("tsql")
-                    && t.get("column_name").equals("n1")
-                    && t.get("data_type").equals("integer")
-                    && t.get("auto_increment").equals(0)
-                    ).findAny().isPresent());
+                        .filter(t
+                                -> t.get("table_name").equals("tsql")
+                        && t.get("column_name").equals("n1")
+                        && t.get("data_type").equals("integer")
+                        && t.get("auto_increment").equals(0)
+                        ).findAny().isPresent());
                 assertTrue(records.stream()
-                    .filter(t
-                        -> t.get("table_name").equals("tsql2")
-                    && t.get("column_name").equals("s1")
-                    && t.get("data_type").equals("timestamp")
-                    ).findAny().isPresent());
+                        .filter(t
+                                -> t.get("table_name").equals("tsql2")
+                        && t.get("column_name").equals("s1")
+                        && t.get("data_type").equals("timestamp")
+                        ).findAny().isPresent());
                 assertTrue(records.stream()
-                    .filter(t
-                        -> t.get("table_name").equals("tsql2")
-                    && t.get("column_name").equals("b1")
-                    && t.get("data_type").equals("bytearray")
-                    ).findAny().isPresent());
+                        .filter(t
+                                -> t.get("table_name").equals("tsql2")
+                        && t.get("column_name").equals("b1")
+                        && t.get("data_type").equals("bytearray")
+                        ).findAny().isPresent());
                 assertTrue(records.stream()
-                    .filter(t
-                        -> t.get("table_name").equals("tsql2")
-                    && t.get("column_name").equals("n1")
-                    && t.get("data_type").equals("long")
-                    ).findAny().isPresent());
+                        .filter(t
+                                -> t.get("table_name").equals("tsql2")
+                        && t.get("column_name").equals("n1")
+                        && t.get("data_type").equals("long")
+                        ).findAny().isPresent());
             }
 
         }
diff --git a/herddb-jdbc/src/test/java/herddb/jdbc/SystemTablesTest.java b/herddb-jdbc/src/test/java/herddb/jdbc/SystemTablesTest.java
index a0ecc72e..15553c6d 100644
--- a/herddb-jdbc/src/test/java/herddb/jdbc/SystemTablesTest.java
+++ b/herddb-jdbc/src/test/java/herddb/jdbc/SystemTablesTest.java
@@ -58,9 +58,10 @@ public class SystemTablesTest {
             try (HDBClient client = new HDBClient(new ClientConfiguration(folder.newFolder().toPath()));) {
                 client.setClientSideMetadataProvider(new StaticClientSideMetadataProvider(server));
                 try (BasicHerdDBDataSource dataSource = new BasicHerdDBDataSource(client);
-                    Connection con = dataSource.getConnection();
-                    Statement statement = con.createStatement();) {
+                        Connection con = dataSource.getConnection();
+                        Statement statement = con.createStatement();) {
                     statement.execute("CREATE TABLE mytable (key string primary key, name string)");
+                    statement.execute("CREATE INDEX mytableindex ON mytable(name)");
                     statement.execute("CREATE TABLE mytable2 (n2 int primary key auto_increment, name string, ts timestamp)");
 
                     try (ResultSet rs = statement.executeQuery("SELECT * FROM SYSTABLES")) {
@@ -129,21 +130,60 @@ public class SystemTablesTest {
                                 String value = rs.getString(i + 1);
                                 record.add(value);
                             }
-                            System.out.println("record:" + record);
                             records.add(record);
                         }
                         assertEquals(3, records.size());
                         assertTrue(records.stream().filter(s -> s.contains("n2")
-                            && s.contains("integer")
+                                && s.contains("integer")
                         ).findAny().isPresent());
                         assertTrue(records.stream().filter(s
-                            -> s.contains("name") && s.contains("string")
+                                -> s.contains("name") && s.contains("string")
                         ).findAny().isPresent());
                         assertTrue(records.stream().filter(s
-                            -> s.contains("ts") && s.contains("timestamp")
+                                -> s.contains("ts") && s.contains("timestamp")
                         ).findAny().isPresent());
                     }
 
+                    try (ResultSet rs = metaData.getIndexInfo(null, null, "mytable", false, false)) {
+
+                        List<List<String>> records = new ArrayList<>();
+                        while (rs.next()) {
+                            List<String> record = new ArrayList<>();
+                            for (int i = 0; i < rs.getMetaData().getColumnCount(); i++) {
+                                String value = rs.getString(i + 1);
+                                record.add(value);
+                            }
+                            records.add(record);
+                        }
+                        assertEquals(2, records.size()); // pk + secondary index
+                    }
+
+                    try (ResultSet rs = metaData.getIndexInfo(null, null, "mytable", true, false)) {
+
+                        List<List<String>> records = new ArrayList<>();
+                        while (rs.next()) {
+                            List<String> record = new ArrayList<>();
+                            for (int i = 0; i < rs.getMetaData().getColumnCount(); i++) {
+                                String value = rs.getString(i + 1);
+                                record.add(value);
+                            }
+                            records.add(record);
+                        }
+                        assertEquals(1, records.size()); // only pk
+                    }
+
+                    try (ResultSet rs = metaData.getIndexInfo(null, null, null, true, false)) {
+                        List<List<String>> records = new ArrayList<>();
+                        while (rs.next()) {
+                            List<String> record = new ArrayList<>();
+                            for (int i = 0; i < rs.getMetaData().getColumnCount(); i++) {
+                                String value = rs.getString(i + 1);
+                                record.add(value);
+                            }
+                            records.add(record);
+                        }
+                        assertEquals(18, records.size());
+                    }
                     try (ResultSet rs = metaData.getSchemas()) {
                         List<List<String>> records = new ArrayList<>();
                         while (rs.next()) {
@@ -159,7 +199,7 @@ public class SystemTablesTest {
 
                     }
 
-                    try (ResultSet rs = metaData.getSchemas(null, "%"+TableSpace.DEFAULT.substring(2))) {
+                    try (ResultSet rs = metaData.getSchemas(null, "%" + TableSpace.DEFAULT.substring(2))) {
                         List<List<String>> records = new ArrayList<>();
                         while (rs.next()) {
                             List<String> record = new ArrayList<>();