Issue 1415 - DynamoDB store for table names & IDs

example/full/instance.properties

@@ -96,6 +96,10 @@ sleeper.metadata.dynamo.pointintimerecovery=false
 # revision DynamoDB table.
 sleeper.metadata.s3.dynamo.pointintimerecovery=false
 
+# This specifies whether point in time recovery is enabled for the Sleeper table index. This is set on
+# the DynamoDB tables.
+sleeper.tables.index.dynamo.pointintimerecovery=false
+
 # The timeout in minutes for when the table properties provider cache should be cleared, forcing table
 # properties to be reloaded from S3.
 sleeper.table.properties.provider.timeout.minutes=60

java/cdk/src/main/java/sleeper/cdk/SleeperCdkApp.java

@@ -39,6 +39,7 @@
 import sleeper.cdk.stack.S3StateStoreStack;
 import sleeper.cdk.stack.StateStoreStacks;
 import sleeper.cdk.stack.TableDataStack;
+import sleeper.cdk.stack.TableIndexStack;
 import sleeper.cdk.stack.TableMetricsStack;
 import sleeper.cdk.stack.TopicStack;
 import sleeper.cdk.stack.VpcStack;
@@ -124,6 +125,7 @@ public void create() {
 
         // Stack for tables
         dataStack = new TableDataStack(this, "TableData", instanceProperties);
+        new TableIndexStack(this, "TableIndex", instanceProperties);
         stateStoreStacks = new StateStoreStacks(
                 new DynamoDBStateStoreStack(this, "DynamoDBStateStore", instanceProperties),
                 new S3StateStoreStack(this, "S3StateStore", instanceProperties, dataStack));

java/cdk/src/main/java/sleeper/cdk/stack/TableIndexStack.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 2022-2023 Crown Copyright
+ *
+ * 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 sleeper.cdk.stack;
+
+import software.amazon.awscdk.NestedStack;
+import software.amazon.awscdk.RemovalPolicy;
+import software.amazon.awscdk.services.dynamodb.Attribute;
+import software.amazon.awscdk.services.dynamodb.AttributeType;
+import software.amazon.awscdk.services.dynamodb.BillingMode;
+import software.amazon.awscdk.services.dynamodb.ITable;
+import software.amazon.awscdk.services.dynamodb.Table;
+import software.constructs.Construct;
+
+import sleeper.configuration.properties.instance.InstanceProperties;
+import sleeper.table.index.dynamodb.DynamoDBTableIndex;
+
+import static sleeper.cdk.Utils.removalPolicy;
+import static sleeper.configuration.properties.instance.CdkDefinedInstanceProperty.TABLE_ID_INDEX_DYNAMO_TABLENAME;
+import static sleeper.configuration.properties.instance.CdkDefinedInstanceProperty.TABLE_NAME_INDEX_DYNAMO_TABLENAME;
+import static sleeper.configuration.properties.instance.CommonProperty.ID;
+import static sleeper.configuration.properties.instance.CommonProperty.TABLE_INDEX_DYNAMO_POINT_IN_TIME_RECOVERY;
+
+public class TableIndexStack extends NestedStack {
+
+    private final ITable indexByNameDynamoTable;
+    private final ITable indexByIdDynamoTable;
+
+    public TableIndexStack(Construct scope, String id, InstanceProperties instanceProperties) {
+        super(scope, id);
+        String instanceId = instanceProperties.get(ID);
+        RemovalPolicy removalPolicy = removalPolicy(instanceProperties);
+
+        indexByNameDynamoTable = Table.Builder
+                .create(this, "IndexByNameDynamoDBTable")
+                .tableName(String.join("-", "sleeper", instanceId, "table-index-by-name"))
+                .removalPolicy(removalPolicy)
+                .billingMode(BillingMode.PAY_PER_REQUEST)
+                .partitionKey(Attribute.builder()
+                        .name(DynamoDBTableIndex.TABLE_NAME_FIELD)
+                        .type(AttributeType.STRING)
+                        .build())
+                .pointInTimeRecovery(instanceProperties.getBoolean(TABLE_INDEX_DYNAMO_POINT_IN_TIME_RECOVERY))
+                .build();
+        instanceProperties.set(TABLE_NAME_INDEX_DYNAMO_TABLENAME, indexByNameDynamoTable.getTableName());
+
+        indexByIdDynamoTable = Table.Builder
+                .create(this, "IndexByIdDynamoDBTable")
+                .tableName(String.join("-", "sleeper", instanceId, "table-index-by-id"))
+                .removalPolicy(removalPolicy)
+                .billingMode(BillingMode.PAY_PER_REQUEST)
+                .partitionKey(Attribute.builder()
+                        .name(DynamoDBTableIndex.TABLE_ID_FIELD)
+                        .type(AttributeType.STRING)
+                        .build())
+                .pointInTimeRecovery(instanceProperties.getBoolean(TABLE_INDEX_DYNAMO_POINT_IN_TIME_RECOVERY))
+                .build();
+        instanceProperties.set(TABLE_ID_INDEX_DYNAMO_TABLENAME, indexByIdDynamoTable.getTableName());
+    }
+
+    public ITable getIndexByNameDynamoTable() {
+        return indexByNameDynamoTable;
+    }
+
+    public ITable getIndexByIdDynamoTable() {
+        return indexByIdDynamoTable;
+    }
+}

java/configuration/src/main/java/sleeper/configuration/properties/instance/CdkDefinedInstanceProperty.java

@@ -46,6 +46,16 @@ public interface CdkDefinedInstanceProperty extends InstanceProperty {
             .propertyGroup(InstancePropertyGroup.COMMON)
             .build();
 
+    // Tables
+    CdkDefinedInstanceProperty TABLE_NAME_INDEX_DYNAMO_TABLENAME = Index.propertyBuilder("sleeper.tables.name.index.dynamo.table")
+            .description("The name of the DynamoDB table indexing tables by their name.")
+            .propertyGroup(InstancePropertyGroup.COMMON)
+            .build();
+    CdkDefinedInstanceProperty TABLE_ID_INDEX_DYNAMO_TABLENAME = Index.propertyBuilder("sleeper.tables.id.index.dynamo.table")
+            .description("The name of the DynamoDB table indexing tables by their ID.")
+            .propertyGroup(InstancePropertyGroup.COMMON)
+            .build();
+
     // DynamoDBStateStore
     CdkDefinedInstanceProperty ACTIVE_FILEINFO_TABLENAME = Index.propertyBuilder("sleeper.metadata.dynamo.active.table")
             .description("The name of the DynamoDB table holding metadata of active files in Sleeper tables.")

java/configuration/src/main/java/sleeper/configuration/properties/instance/CommonProperty.java

@@ -173,6 +173,13 @@ public interface CommonProperty {
             .validationPredicate(Utils::isTrueOrFalse)
             .propertyGroup(InstancePropertyGroup.COMMON)
             .runCdkDeployWhenChanged(true).build();
+    UserDefinedInstanceProperty TABLE_INDEX_DYNAMO_POINT_IN_TIME_RECOVERY = Index.propertyBuilder("sleeper.tables.index.dynamo.pointintimerecovery")
+            .description("This specifies whether point in time recovery is enabled for the Sleeper table index. " +
+                    "This is set on the DynamoDB tables.")
+            .defaultValue("false")
+            .validationPredicate(Utils::isTrueOrFalse)
+            .propertyGroup(InstancePropertyGroup.COMMON)
+            .runCdkDeployWhenChanged(true).build();
 
     UserDefinedInstanceProperty TABLE_PROPERTIES_PROVIDER_TIMEOUT_IN_MINS = Index.propertyBuilder("sleeper.table.properties.provider.timeout.minutes")
             .description("The timeout in minutes for when the table properties provider cache should be cleared, " +

java/configuration/src/test/java/sleeper/configuration/properties/InstancePropertiesTestHelper.java

@@ -32,6 +32,8 @@
 import static sleeper.configuration.properties.instance.CdkDefinedInstanceProperty.PARTITION_TABLENAME;
 import static sleeper.configuration.properties.instance.CdkDefinedInstanceProperty.READY_FOR_GC_FILEINFO_TABLENAME;
 import static sleeper.configuration.properties.instance.CdkDefinedInstanceProperty.REVISION_TABLENAME;
+import static sleeper.configuration.properties.instance.CdkDefinedInstanceProperty.TABLE_ID_INDEX_DYNAMO_TABLENAME;
+import static sleeper.configuration.properties.instance.CdkDefinedInstanceProperty.TABLE_NAME_INDEX_DYNAMO_TABLENAME;
 import static sleeper.configuration.properties.instance.CdkDefinedInstanceProperty.VERSION;
 import static sleeper.configuration.properties.instance.CommonProperty.ACCOUNT;
 import static sleeper.configuration.properties.instance.CommonProperty.ID;
@@ -80,6 +82,8 @@ public static InstanceProperties createTestInstanceProperties() {
         instanceProperties.set(READY_FOR_GC_FILEINFO_TABLENAME, id + "-rfgcf");
         instanceProperties.set(PARTITION_TABLENAME, id + "-p");
         instanceProperties.set(REVISION_TABLENAME, id + "-rv");
+        instanceProperties.set(TABLE_NAME_INDEX_DYNAMO_TABLENAME, id + "-tni");
+        instanceProperties.set(TABLE_ID_INDEX_DYNAMO_TABLENAME, id + "-tii");
         return instanceProperties;
     }
 

java/core/src/main/java/sleeper/core/table/TableAlreadyExistsException.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2022-2023 Crown Copyright
+ *
+ * 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 sleeper.core.table;
+
+public class TableAlreadyExistsException extends RuntimeException {
+
+    public TableAlreadyExistsException(String tableName) {
+        super("Table already exists: " + tableName);
+    }
+}

java/core/src/main/java/sleeper/core/table/TableId.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2022-2023 Crown Copyright
+ *
+ * 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 sleeper.core.table;
+
+import java.util.Objects;
+
+public class TableId {
+
+    private final String tableUniqueId;
+    private final String tableName;
+
+    private TableId(String tableUniqueId, String tableName) {
+        this.tableUniqueId = tableUniqueId;
+        this.tableName = tableName;
+    }
+
+    public static TableId uniqueIdAndName(String tableUniqueId, String tableName) {
+        return new TableId(tableUniqueId, tableName);
+    }
+
+    public String getTableName() {
+        return tableName;
+    }
+
+    public String getTableUniqueId() {
+        return tableUniqueId;
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        if (this == object) {
+            return true;
+        }
+        if (object == null || getClass() != object.getClass()) {
+            return false;
+        }
+        TableId tableId1 = (TableId) object;
+        return Objects.equals(tableUniqueId, tableId1.tableUniqueId) && Objects.equals(tableName, tableId1.tableName);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(tableUniqueId, tableName);
+    }
+
+    @Override
+    public String toString() {
+        return "TableId{" +
+                "tableUniqueId='" + tableUniqueId + '\'' +
+                ", tableName='" + tableName + '\'' +
+                '}';
+    }
+}

java/core/src/main/java/sleeper/core/table/TableIdGenerator.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2022-2023 Crown Copyright
+ *
+ * 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 sleeper.core.table;
+
+import org.apache.commons.codec.binary.Hex;
+
+import java.security.SecureRandom;
+import java.util.Random;
+
+public class TableIdGenerator {
+
+    private final Random random;
+
+    public TableIdGenerator() {
+        random = new SecureRandom();
+    }
+
+    private TableIdGenerator(int seed) {
+        random = new Random(seed);
+    }
+
+    public static TableIdGenerator fromRandomSeed(int seed) {
+        return new TableIdGenerator(seed);
+    }
+
+    public String generateString() {
+        byte[] bytes = new byte[4];
+        random.nextBytes(bytes);
+        return Hex.encodeHexString(bytes);
+    }
+}

java/core/src/main/java/sleeper/core/table/TableIndex.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2022-2023 Crown Copyright
+ *
+ * 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 sleeper.core.table;
+
+import java.util.Optional;
+import java.util.stream.Stream;
+
+public interface TableIndex {
+    TableId createTable(String tableName) throws TableAlreadyExistsException;
+
+    Stream<TableId> streamAllTables();
+
+    Optional<TableId> getTableByName(String tableName);
+
+    Optional<TableId> getTableByUniqueId(String tableUniqueId);
+}

java/core/src/test/java/sleeper/core/table/InMemoryTableIndex.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2022-2023 Crown Copyright
+ *
+ * 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 sleeper.core.table;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.TreeMap;
+import java.util.stream.Stream;
+
+public class InMemoryTableIndex implements TableIndex {
+
+    private final Map<String, TableId> idByName = new TreeMap<>();
+    private final Map<String, TableId> nameById = new HashMap<>();
+    private int nextIdNumber = 1;
+
+    @Override
+    public TableId createTable(String tableName) {
+        if (idByName.containsKey(tableName)) {
+            throw new TableAlreadyExistsException(tableName);
+        }
+        TableId id = TableId.uniqueIdAndName("table-" + nextIdNumber, tableName);
+        nextIdNumber++;
+        idByName.put(tableName, id);
+        nameById.put(id.getTableUniqueId(), id);
+        return id;
+    }
+
+    @Override
+    public Stream<TableId> streamAllTables() {
+        return idByName.values().stream();
+    }
+
+    @Override
+    public Optional<TableId> getTableByName(String tableName) {
+        return Optional.ofNullable(idByName.get(tableName));
+    }
+
+    @Override
+    public Optional<TableId> getTableByUniqueId(String tableUniqueId) {
+        return Optional.ofNullable(nameById.get(tableUniqueId));
+    }
+}