[PIXELS-437] update metadata service. (#71)

Use the new metadata service interface and support specifying storage paths in the create table statement.
pixelsdb · Jun 12, 2023 · 4a34d77 · 4a34d77
1 parent f83b0b2
commit 4a34d77
Show file tree

Hide file tree

Showing 6 changed files with 111 additions and 75 deletions.
diff --git a/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsMetadata.java b/connector/src/main/java/io/pixelsdb/pixels/trino/PixelsMetadata.java
@@ -59,6 +59,8 @@
 import static com.google.common.base.Verify.verify;
 import static com.google.common.base.Verify.verifyNotNull;
 import static com.google.common.collect.ImmutableList.toImmutableList;
+import static io.pixelsdb.pixels.trino.properties.PixelsTableProperties.PATHS;
+import static io.pixelsdb.pixels.trino.properties.PixelsTableProperties.STORAGE;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.toList;
 
@@ -306,12 +308,41 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
         SchemaTableName schemaTableName = tableMetadata.getTable();
         String schemaName = schemaTableName.getSchemaName();
         String tableName = schemaTableName.getTableName();
-        String storageScheme = ((String) Optional.ofNullable(tableMetadata.getProperties().get("storage"))
-                .orElse("hdfs")).toLowerCase(); // use HDFS by default.
-        if (!Storage.Scheme.isValid(storageScheme))
+        String storage = (String) tableMetadata.getProperties().get(STORAGE);
+        String paths = (String) tableMetadata.getProperties().get(PATHS);
+        if (storage == null)
         {
-            throw new TrinoException(PixelsErrorCode.PIXELS_METASTORE_ERROR,
-                    "Unsupported storage scheme '" + storageScheme + "'.");
+            throw new TrinoException(PixelsErrorCode.PIXELS_QUERY_PARSING_ERROR,
+                    "Table must be created with the property 'storage'.");
+        }
+        if (paths == null)
+        {
+            throw new TrinoException(PixelsErrorCode.PIXELS_QUERY_PARSING_ERROR,
+                    "Table must be created with the property 'paths'.");
+        }
+        if (!Storage.Scheme.isValid(storage))
+        {
+            throw new TrinoException(PixelsErrorCode.PIXELS_QUERY_PARSING_ERROR,
+                    "Unsupported storage scheme '" + storage + "'.");
+        }
+        Storage.Scheme storageScheme = Storage.Scheme.from(storage);
+        String[] basePathUris = paths.split(";");
+        for (int i = 0; i < basePathUris.length; ++i)
+        {
+            Storage.Scheme scheme = Storage.Scheme.fromPath(basePathUris[i]);
+            if (scheme == null)
+            {
+                basePathUris[i] = storageScheme + "://" + basePathUris[i];
+            }
+            if (scheme != storageScheme)
+            {
+                throw new TrinoException(PixelsErrorCode.PIXELS_QUERY_PARSING_ERROR,
+                        "The storage schemes in 'paths' are inconsistent with 'storage'.");
+            }
+            if (!basePathUris[i].endsWith("/"))
+            {
+                basePathUris[i] = basePathUris[i] + "/";
+            }
         }
         List<Column> columns = new ArrayList<>();
         for (ColumnMetadata columnMetadata : tableMetadata.getColumns())
@@ -327,8 +358,9 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
         }
         try
         {
-            boolean res = this.metadataProxy.createTable(schemaName, tableName, storageScheme, columns);
-            if (res == false && ignoreExisting == false)
+            boolean res = this.metadataProxy.createTable(schemaName, tableName, storageScheme,
+                    Arrays.asList(basePathUris), columns);
+            if (!res && !ignoreExisting)
             {
                 throw  new TrinoException(PixelsErrorCode.PIXELS_SQL_EXECUTE_ERROR,
                         "Table '" + schemaTableName + "' might already exist, failed to create it.");
@@ -349,7 +381,7 @@ public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle
         try
         {
             boolean res = this.metadataProxy.dropTable(schemaName, tableName);
-            if (res == false)
+            if (!res)
             {
                 throw  new TrinoException(PixelsErrorCode.PIXELS_SQL_EXECUTE_ERROR,
                         "Table " + schemaName + "." + tableName + " does not exist.");