Repository: diennea/herddb
Pull request #8 (2016-04-12 15:21:26)

starting zookeeper metadata manager


Patch:
diff --git a/herddb-core/pom.xml b/herddb-core/pom.xml
index fc09339d..7ab1b513 100644
--- a/herddb-core/pom.xml
+++ b/herddb-core/pom.xml
@@ -20,7 +20,36 @@
         <dependency>
             <groupId>com.github.jsqlparser</groupId>
             <artifactId>jsqlparser</artifactId>
-            <version>0.9.5</version>
+            <version>${libs.jsqlparser}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.zookeeper</groupId>
+            <artifactId>zookeeper</artifactId>
+            <version>${libs.zookeeper}</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>jms</artifactId>
+                    <groupId>javax.jms</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>jmxtools</artifactId>
+                    <groupId>com.sun.jdmk</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>jmxri</artifactId>
+                    <groupId>com.sun.jmx</groupId>
+                </exclusion>
+                <exclusion>
+                    <artifactId>slf4j-log4j12</artifactId>
+                    <groupId>org.slf4j</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <version>${libs.curator}</version>            
+            <scope>test</scope>
         </dependency>
     </dependencies>
     <properties>
diff --git a/herddb-core/src/main/java/herddb/file/FileMetadataStorageManager.java b/herddb-core/src/main/java/herddb/file/FileMetadataStorageManager.java
index 7913173d..4e8893f9 100644
--- a/herddb-core/src/main/java/herddb/file/FileMetadataStorageManager.java
+++ b/herddb-core/src/main/java/herddb/file/FileMetadataStorageManager.java
@@ -110,7 +110,7 @@ public class FileMetadataStorageManager extends MetadataStorageManager {
     }
 
     @Override
-    public void updateTableSpace(TableSpace tableSpace) throws DDLException, MetadataStorageManagerException {
+    public void updateTableSpace(TableSpace tableSpace, TableSpace previous) throws DDLException, MetadataStorageManagerException {
         validateTableSpace(tableSpace);
         lock.writeLock().lock();
         try {
@@ -133,7 +133,7 @@ public class FileMetadataStorageManager extends MetadataStorageManager {
                 if (filename.endsWith(".metadata")) {
                     try (InputStream in = Files.newInputStream(p);
                             DataInputStream iin = new DataInputStream(in);) {
-                        TableSpace ts = TableSpace.deserialize(iin);
+                        TableSpace ts = TableSpace.deserialize(iin, 0);
                         if (filename.equals(ts.name + ".metadata")) {
                             tableSpaces.put(ts.name, ts);
                         }
@@ -161,7 +161,7 @@ public class FileMetadataStorageManager extends MetadataStorageManager {
             throw new MetadataStorageManagerException(err);
         }
     }
-    
+
     @Override
     public void ensureDefaultTableSpace(String localNodeId) throws MetadataStorageManagerException {
         lock.writeLock().lock();
diff --git a/herddb-core/src/main/java/herddb/mem/MemoryMetadataStorageManager.java b/herddb-core/src/main/java/herddb/mem/MemoryMetadataStorageManager.java
index 21a0439f..4982c682 100644
--- a/herddb-core/src/main/java/herddb/mem/MemoryMetadataStorageManager.java
+++ b/herddb-core/src/main/java/herddb/mem/MemoryMetadataStorageManager.java
@@ -75,7 +75,7 @@ public class MemoryMetadataStorageManager extends MetadataStorageManager {
     }
 
     @Override
-    public void updateTableSpace(TableSpace tableSpace) throws DDLException {
+    public void updateTableSpace(TableSpace tableSpace, TableSpace previous) throws DDLException {
         validateTableSpace(tableSpace);
         lock.writeLock().lock();
         try {
diff --git a/herddb-core/src/main/java/herddb/metadata/MetadataStorageManager.java b/herddb-core/src/main/java/herddb/metadata/MetadataStorageManager.java
index 53ac6a22..187795e0 100644
--- a/herddb-core/src/main/java/herddb/metadata/MetadataStorageManager.java
+++ b/herddb-core/src/main/java/herddb/metadata/MetadataStorageManager.java
@@ -65,7 +65,7 @@ public abstract class MetadataStorageManager implements AutoCloseable {
      *
      * @param tableSpace
      */
-    public abstract void updateTableSpace(TableSpace tableSpace) throws DDLException, MetadataStorageManagerException;
+    public abstract void updateTableSpace(TableSpace tableSpace, TableSpace previous) throws DDLException, MetadataStorageManagerException;
 
     protected void validateTableSpace(TableSpace tableSpace) throws DDLException {
         // TODO: implement sensible validations
diff --git a/herddb-core/src/main/java/herddb/metadata/MetadataStorageManagerException.java b/herddb-core/src/main/java/herddb/metadata/MetadataStorageManagerException.java
index 072768be..db0a74e7 100644
--- a/herddb-core/src/main/java/herddb/metadata/MetadataStorageManagerException.java
+++ b/herddb-core/src/main/java/herddb/metadata/MetadataStorageManagerException.java
@@ -30,4 +30,8 @@ public class MetadataStorageManagerException extends Exception {
         super(cause);
     }
 
+    public MetadataStorageManagerException(String message) {
+        super(message);
+    }
+
 }
diff --git a/herddb-core/src/main/java/herddb/model/TableSpace.java b/herddb-core/src/main/java/herddb/model/TableSpace.java
index 81cd71b7..f04fb884 100644
--- a/herddb-core/src/main/java/herddb/model/TableSpace.java
+++ b/herddb-core/src/main/java/herddb/model/TableSpace.java
@@ -20,6 +20,8 @@
 package herddb.model;
 
 import herddb.log.LogSequenceNumber;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
@@ -51,18 +53,25 @@ public class TableSpace {
 
     public final LogSequenceNumber lastCheckpointLogPosition;
 
-    private TableSpace(String name, String leaderId, Set<String> replicas, LogSequenceNumber lastCheckpointLogPosition) {
+    public final Object metadataStorageVersion;
+
+    private TableSpace(String name, String leaderId, Set<String> replicas, LogSequenceNumber lastCheckpointLogPosition, Object metadataStorageVersion) {
         this.name = name;
         this.leaderId = leaderId;
         this.replicas = replicas;
         this.lastCheckpointLogPosition = lastCheckpointLogPosition;
+        this.metadataStorageVersion = metadataStorageVersion;
     }
 
     public static Builder builder() {
         return new Builder();
     }
 
-    public static TableSpace deserialize(DataInputStream in) throws IOException {
+    public static TableSpace deserialize(byte[] data, Object metadataStorageVersion) throws IOException {
+        return deserialize(new DataInputStream(new ByteArrayInputStream(data)), metadataStorageVersion);
+    }
+
+    public static TableSpace deserialize(DataInputStream in, Object metadataStorageVersion) throws IOException {
         String name = in.readUTF();
         String leaderId = in.readUTF();
         int numreplicas = in.readInt();
@@ -71,7 +80,15 @@ public class TableSpace {
             replicas.add(in.readUTF());
         }
         LogSequenceNumber number = new LogSequenceNumber(in.readLong(), in.readLong());
-        return new TableSpace(name, leaderId, replicas, number);
+        return new TableSpace(name, leaderId, replicas, number, metadataStorageVersion);
+    }
+
+    public byte[] serialize() throws IOException {
+        ByteArrayOutputStream oo = new ByteArrayOutputStream();
+        try (DataOutputStream doo = new DataOutputStream(oo)) {
+            serialize(doo);
+        }
+        return oo.toByteArray();
     }
 
     public void serialize(DataOutputStream out) throws IOException {
@@ -138,7 +155,7 @@ public class TableSpace {
             if (!replicas.contains(leaderId)) {
                 throw new IllegalArgumentException("leader " + leaderId + " must be in replica list " + replicas);
             }
-            return new TableSpace(name, leaderId, Collections.unmodifiableSet(replicas), lastCheckpointLogPosition);
+            return new TableSpace(name, leaderId, Collections.unmodifiableSet(replicas), lastCheckpointLogPosition, null);
         }
 
     }
diff --git a/herddb-core/src/main/java/herddb/zookeeper/ZookeeperMetadataStorageManager.java b/herddb-core/src/main/java/herddb/zookeeper/ZookeeperMetadataStorageManager.java
new file mode 100644
index 00000000..5d1daf14
--- /dev/null
+++ b/herddb-core/src/main/java/herddb/zookeeper/ZookeeperMetadataStorageManager.java
@@ -0,0 +1,186 @@
+/*
+ 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.zookeeper;
+
+import herddb.metadata.MetadataStorageManager;
+import herddb.metadata.MetadataStorageManagerException;
+import herddb.model.DDLException;
+import herddb.model.TableSpace;
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+
+/**
+ * Metadata storage manager over Zookeeper
+ *
+ * @author enrico.olivelli
+ */
+public class ZookeeperMetadataStorageManager extends MetadataStorageManager {
+
+    private ZooKeeper zooKeeper;
+    private final String zkAddress;
+    private final int zkSessionTimeout;
+    private final String basePath;
+    private final CountDownLatch firstConnectionLatch = new CountDownLatch(1);
+    private final Watcher mainWatcher = new Watcher() {
+        @Override
+        public void process(WatchedEvent event) {
+            switch (event.getState()) {
+                case SyncConnected:
+                    firstConnectionLatch.countDown();
+                    break;
+            }
+        }
+    };
+
+    public ZookeeperMetadataStorageManager(String zkAddress, int zkSessionTimeout, String basePath) {
+        this.zkAddress = zkAddress;
+        this.zkSessionTimeout = zkSessionTimeout;
+        this.basePath = basePath;
+    }
+
+    @Override
+    public void start() throws MetadataStorageManagerException {
+        try {
+            this.zooKeeper = new ZooKeeper(zkAddress, zkSessionTimeout, mainWatcher);
+            firstConnectionLatch.await(zkSessionTimeout, TimeUnit.SECONDS); // TODO: use another timeout?
+            ensureRoot();
+        } catch (IOException | InterruptedException | KeeperException err) {
+            throw new MetadataStorageManagerException(err);
+        }
+    }
+
+    private void ensureRoot() throws KeeperException, InterruptedException {
+        try {
+            this.zooKeeper.create(basePath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        } catch (KeeperException.NodeExistsException ok) {
+        }
+        try {
+            this.zooKeeper.create(basePath + "/tableSpaces", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+        } catch (KeeperException.NodeExistsException ok) {
+        }
+    }
+
+    private static class TableSpaceList {
+
+        private final int version;
+        private final List<String> tableSpaces;
+
+        public TableSpaceList(int version, List<String> tableSpaces) {
+            this.version = version;
+            this.tableSpaces = tableSpaces;
+        }
+
+    }
+
+    private TableSpaceList listTablesSpaces() throws KeeperException, InterruptedException {
+        Stat stat = new Stat();
+        List<String> children = zooKeeper.getChildren(basePath + "/tableSpaces", false, stat);
+        return new TableSpaceList(stat.getVersion(), children);
+    }
+
+    private void createTableSpaceNode(TableSpace tableSpace) throws KeeperException, InterruptedException, IOException {
+        zooKeeper.create(basePath + "/tableSpaces/" + tableSpace.name, tableSpace.serialize(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+    }
+
+    private void updateTableSpaceNode(TableSpace tableSpace, int metadataStorageVersion) throws KeeperException, InterruptedException, IOException {
+        zooKeeper.setData(basePath + "/tableSpaces/" + tableSpace.name, tableSpace.serialize(), metadataStorageVersion);
+    }
+
+    @Override
+    public void ensureDefaultTableSpace(String localNodeId) throws MetadataStorageManagerException {
+        try {
+            TableSpaceList list = listTablesSpaces();
+            if (!list.tableSpaces.contains(TableSpace.DEFAULT)) {
+                TableSpace tableSpace = TableSpace.builder().leader(localNodeId).replica(localNodeId).name(TableSpace.DEFAULT).build();
+                createTableSpaceNode(tableSpace);
+            }
+        } catch (KeeperException.NodeExistsException err) {
+            // not a problem
+        } catch (InterruptedException | KeeperException | IOException err) {
+            throw new MetadataStorageManagerException(err);
+        }
+    }
+
+    @Override
+    public void close() throws MetadataStorageManagerException {
+        if (this.zooKeeper != null) {
+            try {
+                zooKeeper.close();
+            } catch (InterruptedException err) {
+            }
+        }
+    }
+
+    @Override
+    public Collection<String> listTableSpaces() throws MetadataStorageManagerException {
+        try {
+            return listTablesSpaces().tableSpaces;
+        } catch (KeeperException | InterruptedException ex) {
+            throw new MetadataStorageManagerException(ex);
+        }
+    }
+
+    @Override
+    public TableSpace describeTableSpace(String name) throws MetadataStorageManagerException {
+        try {
+            Stat stat = new Stat();
+            byte[] result = zooKeeper.getData(basePath + "/tableSpaces/" + name, false, stat);
+            return TableSpace.deserialize(result, stat.getVersion());
+        } catch (KeeperException.NoNodeException ex) {
+            return null;
+        } catch (KeeperException | InterruptedException | IOException ex) {
+            throw new MetadataStorageManagerException(ex);
+        }
+    }
+
+    @Override
+    public void registerTableSpace(TableSpace tableSpace) throws DDLException, MetadataStorageManagerException {
+
+        try {
+            createTableSpaceNode(tableSpace);
+        } catch (KeeperException | InterruptedException | IOException ex) {
+            throw new MetadataStorageManagerException(ex);
+        }
+    }
+
+    @Override
+    public void updateTableSpace(TableSpace tableSpace, TableSpace previous) throws DDLException, MetadataStorageManagerException {
+        if (previous.metadataStorageVersion == null) {
+            throw new MetadataStorageManagerException("metadataStorageVersion not read from ZK");
+        }
+        try {
+            updateTableSpaceNode(tableSpace, (Integer) previous.metadataStorageVersion);
+        } catch (KeeperException | InterruptedException | IOException ex) {
+            throw new MetadataStorageManagerException(ex);
+        }
+
+    }
+
+}
diff --git a/pom.xml b/pom.xml
index 14bcdf71..6e5d0649 100644
--- a/pom.xml
+++ b/pom.xml
@@ -54,6 +54,9 @@
         <maven.compiler.target>1.8</maven.compiler.target>
         <libs.netty4>4.0.31.Final</libs.netty4>
         <libs.jackson.mapper>1.9.11</libs.jackson.mapper>
+        <libs.zookeeper>3.4.6</libs.zookeeper>
+        <libs.jsqlparser>0.9.5</libs.jsqlparser>
+        <libs.curator>2.8.0</libs.curator>
     </properties>
         
     <dependencies>


Test patch:
diff --git a/herddb-core/src/test/java/herddb/core/BaseTestcase.java b/herddb-core/src/test/java/herddb/core/BaseTestcase.java
index 3254f723..9b16e18b 100644
--- a/herddb-core/src/test/java/herddb/core/BaseTestcase.java
+++ b/herddb-core/src/test/java/herddb/core/BaseTestcase.java
@@ -50,6 +50,11 @@ public class BaseTestcase {
     protected MetadataStorageManager metadataStorageManager;
     protected CommitLogManager commitLogManager;
 
+    protected void beforeSetup() throws Exception {    
+    }
+    protected void afterTeardown() throws Exception {    
+    }
+
     protected CommitLogManager makeCommitLogManager() throws Exception {
         return new MemoryCommitLogManager();
     }
@@ -64,6 +69,7 @@ public class BaseTestcase {
 
     @Before
     public void setup() throws Exception {
+        beforeSetup();
         metadataStorageManager = makeMetadataStorageManager();
         commitLogManager = makeCommitLogManager();
         dataStorageManager = makeDataStorageManager();
@@ -99,6 +105,6 @@ public class BaseTestcase {
         dataStorageManager = null;
         metadataStorageManager = null;
         commitLogManager = null;
-
+        afterTeardown();
     }
 }
diff --git a/herddb-core/src/test/java/herddb/core/SimpleDMLZookeeperTest.java b/herddb-core/src/test/java/herddb/core/SimpleDMLZookeeperTest.java
new file mode 100644
index 00000000..033c3181
--- /dev/null
+++ b/herddb-core/src/test/java/herddb/core/SimpleDMLZookeeperTest.java
@@ -0,0 +1,162 @@
+/*
+ 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;
+
+import herddb.file.FileCommitLogManager;
+import herddb.log.CommitLogManager;
+import herddb.mem.MemoryDataStorageManager;
+import herddb.metadata.MetadataStorageManager;
+import herddb.model.GetResult;
+import herddb.model.commands.InsertStatement;
+import herddb.model.Record;
+import herddb.model.commands.DeleteStatement;
+import herddb.model.commands.GetStatement;
+import herddb.model.commands.UpdateStatement;
+import herddb.utils.Bytes;
+import herddb.zookeeper.ZookeeperMetadataStorageManager;
+import java.util.List;
+import org.apache.curator.test.TestingServer;
+import org.junit.After;
+import org.junit.Test;
+import org.junit.Rule;
+import org.junit.rules.TemporaryFolder;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import org.junit.Before;
+
+/**
+ * Tests on table creation
+ *
+ * @author enrico.olivelli
+ */
+public class SimpleDMLZookeeperTest extends BaseTestcase {
+
+    @Rule
+    public TemporaryFolder folder = new TemporaryFolder();
+    private TestingServer testingServer;
+
+    @Override
+    protected CommitLogManager makeCommitLogManager() throws Exception {
+        return new FileCommitLogManager(folder.newFolder().toPath());
+    }
+
+    @Override
+    protected void beforeSetup() throws Exception {
+        testingServer = new TestingServer();
+        testingServer.start();
+    }
+
+    @Override
+    public void afterTeardown() throws Exception {
+        if (testingServer != null) {
+            testingServer.stop();
+        }
+    }
+
+    @Override
+    protected MetadataStorageManager makeMetadataStorageManager() throws Exception {
+        return new ZookeeperMetadataStorageManager(testingServer.getConnectString(), 60, "/tests");
+    }
+
+    @Test
+    public void test() throws Exception {
+
+        {
+            Record record = new Record(Bytes.from_string("key1"), Bytes.from_string("0"));
+            InsertStatement st = new InsertStatement(tableSpace, tableName, record);
+            assertEquals(1, manager.executeUpdate(st).getUpdateCount());
+        }
+        assertNull(dataStorageManager.loadPage(tableName, 1L));
+        assertEquals(0, dataStorageManager.getActualNumberOfPages(tableName));
+        manager.flush();
+        assertNotNull(dataStorageManager.loadPage(tableName, 1L));
+        assertEquals(1, dataStorageManager.getActualNumberOfPages(tableName));
+
+        {
+            GetResult result = manager.get(new GetStatement(tableSpace, tableName, Bytes.from_string("key1"), null));
+            assertTrue(result.found());
+        }
+
+        manager.flush();
+        assertEquals(1, dataStorageManager.getActualNumberOfPages(tableName));
+
+        {
+            Record record = new Record(Bytes.from_string("key1"), Bytes.from_string("5"));
+            UpdateStatement st = new UpdateStatement(tableSpace, tableName, record, null);
+            assertEquals(1, manager.executeUpdate(st).getUpdateCount());
+        }
+
+        // a new page must be allocated
+        manager.flush();
+        assertEquals(2, dataStorageManager.getActualNumberOfPages(tableName));
+
+        {
+            Record record = new Record(Bytes.from_string("key1"), Bytes.from_string("6"));
+            UpdateStatement st = new UpdateStatement(tableSpace, tableName, record, null);
+            assertEquals(1, manager.executeUpdate(st).getUpdateCount());
+        }
+        {
+            Record record = new Record(Bytes.from_string("key1"), Bytes.from_string("7"));
+            UpdateStatement st = new UpdateStatement(tableSpace, tableName, record, null);
+            assertEquals(1, manager.executeUpdate(st).getUpdateCount());
+        }
+        // only a new page must be allocated, not two more
+        manager.flush();
+        assertEquals(3, dataStorageManager.getActualNumberOfPages(tableName));
+
+        {
+            DeleteStatement st = new DeleteStatement(tableSpace, tableName, Bytes.from_string("key1"), null);
+            assertEquals(1, manager.executeUpdate(st).getUpdateCount());
+            GetResult result = manager.get(new GetStatement(tableSpace, tableName, Bytes.from_string("key1"), null));
+            assertFalse(result.found());
+        }
+
+        // a delete does not trigger new pages in this case
+        manager.flush();
+        assertEquals(3, dataStorageManager.getActualNumberOfPages(tableName));
+
+        {
+            assertEquals(1, manager.executeUpdate(new InsertStatement(tableSpace, tableName, new Record(Bytes.from_string("key2"), Bytes.from_string("50")))).getUpdateCount());
+            assertEquals(1, manager.executeUpdate(new InsertStatement(tableSpace, tableName, new Record(Bytes.from_string("key3"), Bytes.from_string("60")))).getUpdateCount());
+        }
+
+        manager.flush();
+        assertEquals(4, dataStorageManager.getActualNumberOfPages(tableName));
+        {
+            DeleteStatement st = new DeleteStatement(tableSpace, tableName, Bytes.from_string("key2"), null);
+            assertEquals(1, manager.executeUpdate(st).getUpdateCount());
+        }
+        // a new page, containg the key3 record is needed
+        manager.flush();
+
+        MemoryDataStorageManager mem = (MemoryDataStorageManager) dataStorageManager;
+        for (long pageId = 1; pageId <= dataStorageManager.getActualNumberOfPages(tableName); pageId++) {
+            MemoryDataStorageManager.Page page = mem.getPage(tableName, pageId);
+            List<Record> records = page.getRecords();
+            System.out.println("PAGE #" + pageId + " records :" + records + " seq " + page.getSequenceNumber());
+        }
+
+        assertEquals(5, dataStorageManager.getActualNumberOfPages(tableName));
+
+    }
+}