feat(sql): implement "touch" function, useful to preload the data from disk...

feat(sql): implement "touch" function, useful to preload the data from disk into system page cache. (#1305)

core/src/main/java/io/questdb/griffin/FunctionParser.java

@@ -179,11 +179,11 @@ public class FunctionParser implements PostOrderTreeTraversalAlgo.Visitor {
     }
 
     public boolean isCursor(CharSequence token) {
-        return functionFactoryCache.isCursor(token);
+        return token != null && functionFactoryCache.isCursor(token);
     }
 
     public boolean isGroupBy(CharSequence token) {
-        return functionFactoryCache.isGroupBy(token);
+        return token != null && functionFactoryCache.isGroupBy(token);
     }
 
     public boolean isRuntimeConstant(CharSequence token) {

core/src/main/java/io/questdb/griffin/engine/functions/table/TouchTableFunctionFactory.java

+/*******************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2020 QuestDB
+ *
+ *  Licensed 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 io.questdb.griffin.engine.functions.table;
+
+import io.questdb.TelemetryJob;
+import io.questdb.cairo.BitmapIndexReader;
+import io.questdb.cairo.CairoConfiguration;
+import io.questdb.cairo.sql.*;
+import io.questdb.griffin.FunctionFactory;
+import io.questdb.griffin.SqlException;
+import io.questdb.griffin.SqlExecutionContext;
+import io.questdb.griffin.engine.functions.StrFunction;
+import io.questdb.griffin.engine.functions.UnaryFunction;
+import io.questdb.log.Log;
+import io.questdb.log.LogFactory;
+import io.questdb.std.*;
+import io.questdb.std.str.CharSink;
+import io.questdb.std.str.StringSink;
+
+public class TouchTableFunctionFactory implements FunctionFactory {
+    @Override
+    public String getSignature() {
+        return "touch(C)";
+    }
+
+    @Override
+    public Function newInstance(int position,
+                                ObjList<Function> args,
+                                IntList argPositions,
+                                CairoConfiguration configuration,
+                                SqlExecutionContext sqlExecutionContext
+    ) throws SqlException {
+        final Function function = args.get(0);
+        final int pos = argPositions.get(0);
+
+        try(final RecordCursorFactory recordCursorFactory = function.getRecordCursorFactory()) {
+            if (recordCursorFactory == null || !recordCursorFactory.supportPageFrameCursor()) {
+                throw SqlException.$(pos, "query does not support framing execution and cannot be pre-touched");
+            }
+        }
+
+        return new TouchTableFunc(function);
+    }
+
+    private static class TouchTableFunc extends StrFunction implements UnaryFunction {
+        private static final Log LOG = LogFactory.getLog(TouchTableFunc.class);
+
+        private final Function arg;
+
+        private SqlExecutionContext sqlExecutionContext;
+
+        private final StringSink sinkA = new StringSink();
+        private final StringSink sinkB = new StringSink();
+
+        private long garbage = 0;
+        private long dataPages = 0;
+        private long indexKeyPages = 0;
+        private long indexValuePages = 0;
+
+        public TouchTableFunc(Function arg) {
+            this.arg = arg;
+        }
+
+        @Override
+        public Function getArg() {
+            return arg;
+        }
+
+        @Override
+        public void init(SymbolTableSource symbolTableSource, SqlExecutionContext executionContext) throws SqlException {
+            arg.init(symbolTableSource, executionContext);
+            this.sqlExecutionContext = executionContext;
+        }
+
+        @Override
+        public CharSequence getStr(Record rec) {
+            sinkA.clear();
+            getStr(rec, sinkA);
+            return sinkA;
+        }
+
+        @Override
+        public CharSequence getStrB(Record rec) {
+            sinkB.clear();
+            getStr(rec, sinkB);
+            return sinkB;
+        }
+
+        @Override
+        public void getStr(Record rec, CharSink sink) {
+            touchTable();
+            sink.put("{\"data_pages\": ")
+                    .put(dataPages)
+                    .put(", \"index_key_pages\":")
+                    .put(indexKeyPages)
+                    .put( ", \"index_values_pages\": ")
+                    .put(indexValuePages).put("}");
+        }
+
+        private void clearCounters() {
+            garbage = 0;
+            dataPages = 0;
+            indexKeyPages = 0;
+            indexValuePages = 0;
+        }
+
+        private void touchTable() {
+            clearCounters();
+            final long pageSize = Files.PAGE_SIZE;
+            try (RecordCursorFactory recordCursorFactory = arg.getRecordCursorFactory()) {
+                try (PageFrameCursor pageFrameCursor = recordCursorFactory.getPageFrameCursor(sqlExecutionContext)) {
+                    PageFrame frame;
+                    RecordMetadata metadata = recordCursorFactory.getMetadata();
+                    while ((frame = pageFrameCursor.next()) != null) {
+                        for (int columnIndex = 0, sz = metadata.getColumnCount(); columnIndex < sz; columnIndex++) {
+
+                            final long columnMemorySize = frame.getPageSize(columnIndex);
+                            final long columnBaseAddress = frame.getPageAddress(columnIndex);
+                            dataPages += touchMemory(pageSize, columnBaseAddress, columnMemorySize);
+
+                            if (metadata.isColumnIndexed(columnIndex)) {
+                                final BitmapIndexReader indexReader = frame.getBitmapIndexReader(columnIndex, BitmapIndexReader.DIR_BACKWARD);
+
+                                final long keyBaseAddress = indexReader.getKeyBaseAddress();
+                                final long keyMemorySize = indexReader.getKeyMemorySize();
+                                indexKeyPages += touchMemory(pageSize, keyBaseAddress, keyMemorySize);
+
+                                final long valueBaseAddress = indexReader.getValueBaseAddress();
+                                final long valueMemorySize = indexReader.getValueMemorySize();
+                                indexValuePages += touchMemory(pageSize, valueBaseAddress, valueMemorySize);
+                            }
+                        }
+                    }
+                } catch (SqlException e) {
+                    // do not propagate
+                    LOG.error().$("cannot acquire page frame cursor: ").$((Sinkable) e).$();
+                }
+            }
+        }
+
+        private long touchMemory(long pageSize, long baseAddress, long memorySize) {
+            final long pageCount = (memorySize + pageSize - 1) / pageSize;
+
+            for (long i = 0; i < pageCount; i++) {
+                final byte v = Unsafe.getUnsafe().getByte(baseAddress + i * pageSize);
+                garbage += v;
+            }
+
+            return pageCount;
+        }
+    }
+}
\ No newline at end of file

core/src/main/java/module-info.java

@@ -549,6 +549,7 @@ open module io.questdb {
             io.questdb.griffin.engine.functions.math.PowDoubleFunctionFactory,
             io.questdb.griffin.engine.functions.table.AllTablesFunctionFactory,
             io.questdb.griffin.engine.functions.table.TableColumnsFunctionFactory,
+            io.questdb.griffin.engine.functions.table.TouchTableFunctionFactory,
             // first
             io.questdb.griffin.engine.functions.groupby.FirstSymbolGroupByFunctionFactory,
 //                  Change string case

core/src/main/resources/META-INF/services/io.questdb.griffin.FunctionFactory

@@ -507,6 +507,7 @@ io.questdb.griffin.engine.functions.groupby.AvgDoubleGroupByFunctionFactory
 io.questdb.griffin.engine.functions.math.PowDoubleFunctionFactory
 io.questdb.griffin.engine.functions.table.AllTablesFunctionFactory
 io.questdb.griffin.engine.functions.table.TableColumnsFunctionFactory
+io.questdb.griffin.engine.functions.table.TouchTableFunctionFactory
 
 io.questdb.griffin.engine.functions.groupby.FirstSymbolGroupByFunctionFactory
 

core/src/test/java/io/questdb/griffin/engine/functions/TouchTableFunctionTest.java

+/*******************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2020 QuestDB
+ *
+ *  Licensed 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 io.questdb.griffin.engine.functions;
+
+import io.questdb.griffin.AbstractGriffinTest;
+import io.questdb.griffin.SqlException;
+import io.questdb.griffin.engine.functions.rnd.SharedRandom;
+import io.questdb.std.Rnd;
+import io.questdb.test.tools.TestUtils;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TouchTableFunctionTest extends AbstractGriffinTest {
+    final static String ddl = "create table x as " +
+            "(" +
+            "select" +
+            " rnd_geohash(40) g," +
+            " rnd_double(0)*100 a," +
+            " rnd_symbol(5,4,4,1) b," +
+            " timestamp_sequence(0, 100000000000) k" +
+            " from long_sequence(20)" +
+            "), index(b) timestamp(k) partition by DAY";
+
+    @Before
+    public void setUp3() {
+        SharedRandom.RANDOM.set(new Rnd());
+    }
+
+    @Test
+    public void testTouchTableNoTimestampColumnSelected() throws Exception {
+        assertMemoryLeak(() -> {
+            final String query = "select touch(select g,a,b from x where k in '1970-01-22')";
+            try {
+                execQuery(ddl, query);
+            } catch (SqlException ex) {
+                TestUtils.assertContains(ex.getFlyweightMessage(), "query does not support framing execution and cannot be pre-touched");
+            }
+        });
+    }
+
+    @Test
+    public void testTouchTableThrowOnComplexFilter() throws Exception {
+        assertMemoryLeak(() -> {
+            final String query = "select touch(select * from x where k in '1970-01-22' and a > 100.0)";
+            try {
+                execQuery(ddl, query);
+            } catch (SqlException ex) {
+                TestUtils.assertContains(ex.getFlyweightMessage(), "query does not support framing execution and cannot be pre-touched");
+            }
+        });
+    }
+
+    @Test
+    public void testTouchTableTimeInterval() throws Exception {
+        assertMemoryLeak(() -> {
+            final String query = "select touch(select * from x where k in '1970-01-22')";
+            try {
+                execQuery(ddl, query);
+            } catch (SqlException ex) {
+                Assert.fail(ex.getMessage());
+            }
+        });
+    }
+
+    @Test
+    public void testTouchTableTimeRange() throws Exception {
+        assertMemoryLeak(() -> {
+            final String query = "select touch(select * from x where k > '1970-01-18T00:00:00.000000Z')";
+            try {
+                execQuery(ddl, query);
+            } catch (SqlException ex) {
+                TestUtils.assertContains(ex.getFlyweightMessage(), "query does not support framing execution and cannot be pre-touched");
+            }
+        });
+    }
+
+    @Test
+    public void testTouchUpdateTouchAgain() throws Exception {
+        assertMemoryLeak(() -> {
+
+            final String query = "select touch(select * from x)";
+
+            final String ddl2 = "insert into x select * from (" +
+                    " select" +
+                    " rnd_geohash(40)," +
+                    " rnd_double(0)*100," +
+                    " 'VTJW'," +
+                    " to_timestamp('2019', 'yyyy') t" +
+                    " from long_sequence(100)" +
+                    ") timestamp (t)";
+
+            try {
+                execQuery(ddl, query);
+                execQuery(ddl2, query);
+            } catch (SqlException ex) {
+                Assert.fail(ex.getMessage());
+            }
+
+        });
+    }
+
+    private void execQuery(String ddl, String query) throws SqlException {
+        compiler.compile(ddl, sqlExecutionContext);
+        TestUtils.printSql(compiler, sqlExecutionContext, query, sink);
+    }
+}