[CDAP-21172] Add Creation/Deletion logic for Metadata Tables

123-komal · web-flow · commit f558b9896cd1 · 2025-06-13T18:06:17.000+05:30
diff --git a/cdap-master/pom.xml b/cdap-master/pom.xml
@@ -113,11 +113,6 @@
       <artifactId>cdap-messaging-ext-spanner</artifactId>
       <version>${project.version}</version>
     </dependency>
-    <dependency>
-      <groupId>io.cdap.cdap</groupId>
-      <artifactId>cdap-metadata-ext-spanner</artifactId>
-      <version>${project.version}</version>
-    </dependency>
     <dependency>
       <groupId>org.apache.tephra</groupId>
       <artifactId>tephra-api</artifactId>
@@ -256,7 +251,6 @@
         <stage.authenticators.ext.dir>${stage.opt.dir}/ext/authenticators</stage.authenticators.ext.dir>
         <stage.credential.provider.extensions.dir>${stage.opt.dir}/ext/credentialproviders</stage.credential.provider.extensions.dir>
         <stage.messaging.ext.dir>${stage.opt.dir}/ext/messagingproviders</stage.messaging.ext.dir>
-        <stage.metadata.ext.dir>${stage.opt.dir}/ext/metadatastorage</stage.metadata.ext.dir>
         <stage.metricswriters.ext.dir>${stage.opt.dir}/ext/metricswriters/google_cloud_monitoring_writer
         </stage.metricswriters.ext.dir>
         <stage.eventwriters.ext.dir>${stage.opt.dir}/ext/eventwriters/google_cloud_pubsub_writer
@@ -739,28 +733,6 @@
                 </configuration>
               </execution>
 
-              <!-- Copy metadata extensions -->
-              <execution>
-                <id>copy-metadata-ext-spanner</id>
-                <phase>process-resources</phase>
-                <goals>
-                  <goal>copy-resources</goal>
-                </goals>
-                <configuration combine.self="override">
-                  <outputDirectory>${stage.metadata.ext.dir}/spanner</outputDirectory>
-                  <resources>
-                    <resource>
-                      <directory>
-                        ${project.parent.basedir}/cdap-metadata-ext-spanner/target/libexec/
-                      </directory>
-                      <includes>
-                        <include>*.jar</include>
-                      </includes>
-                    </resource>
-                  </resources>
-                </configuration>
-              </execution>
-
               <!-- Copy remote authenticator extensions -->
               <execution>
                 <id>copy-authenticator-ext-gcp</id>
@@ -848,7 +820,7 @@
                         <exclude name="${additional.artifacts.exclude.pattern}"/>
                         <exclude name="**/target/*-sources.jar"/>
                         <exclude name="**/target/*-javadoc.jar"/>
-                        <!-- Wrangler excludes --> 
+                        <!-- Wrangler excludes -->
                         <exclude name="**/target/wrangler-core*"/>
                         <exclude name="**/target/wrangler-test*"/>
                         <!--We don't want to package the following plugins with CDAP-->
diff --git a/cdap-metadata-ext-spanner/src/main/java/io/cdap/cdap/metadata/spanner/SpannerMetadataStorage.java b/cdap-metadata-ext-spanner/src/main/java/io/cdap/cdap/metadata/spanner/SpannerMetadataStorage.java
@@ -16,6 +16,12 @@
 
 package io.cdap.cdap.metadata.spanner;
 
+import com.google.cloud.spanner.DatabaseAdminClient;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.Spanner;
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerOptions;
+import com.google.common.util.concurrent.Uninterruptibles;
 import io.cdap.cdap.spi.metadata.Metadata;
 import io.cdap.cdap.spi.metadata.MetadataChange;
 import io.cdap.cdap.spi.metadata.MetadataMutation;
@@ -25,36 +31,249 @@
 import io.cdap.cdap.spi.metadata.Read;
 import io.cdap.cdap.spi.metadata.SearchRequest;
 import io.cdap.cdap.spi.metadata.SearchResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
 
 /**
  * A metadata storage provider that delegates to Spanner.
  */
 public class SpannerMetadataStorage implements MetadataStorage {
 
+  private static final Logger LOG = LoggerFactory.getLogger(SpannerMetadataStorage.class);
+
+  // Metadata table names
+  private static final String METADATA_TABLE = "metadata";
+  private static final String METADATA_PROPS_TABLE = "metadata_props";
+
+  public static final class MetadataTable {
+    private static final String METADATA_ID_FIELD = "metadata_id";
+    private static final String METADATA_COLUMN_FIELD = "metadata_column";
+    private static final String NAMESPACE_FIELD = "namespace";
+    private static final String TYPE_FIELD = "entity_type";
+    private static final String NAME_FIELD = "name";
+    private static final String VERSION = "version";
+    private static final String CREATED_FIELD = "create_time";
+    private static final String USER_FIELD = "user";
+    private static final String SYSTEM_FIELD = "system";
+  }
+
+  public static final class MetadataPropsTable {
+    private static final String METADATA_ID_FIELD = "metadata_id";
+    private static final String NAMESPACE_FIELD = "namespace";
+    private static final String TYPE_FIELD = "entity_type";
+    private static final String NESTED_NAME_FIELD = "name";
+    private static final String NESTED_SCOPE_FIELD = "scope";
+    private static final String NESTED_VALUE_FIELD = "value";
+  }
+
+  private String instanceId;
+  private String projectId;
+  private String databaseId;
+
+  private Spanner spanner;
+  private DatabaseAdminClient adminClient;
+
   @Override
   public void initialize(MetadataStorageContext context) throws Exception {
-    throw new IOException("NOT IMPLEMENTED");
+    Map<String, String> properties = context.getProperties();
+
+    this.projectId = Objects.requireNonNull(properties.get("project"));
+    this.instanceId = Objects.requireNonNull(properties.get("instance"));
+    this.databaseId = Objects.requireNonNull(properties.get("database"));
+
+    this.spanner = SpannerOptions.newBuilder().setProjectId(projectId).build().getService();
+    this.adminClient = spanner.getDatabaseAdminClient();
+
+    LOG.info("SpannerMetadataStorage initialized.");
   }
 
   @Override
-  public void close() {
+  public String getName() {
+    return "gcp-spanner";
   }
 
   @Override
   public void createIndex() throws IOException {
-    throw new IOException("NOT IMPLEMENTED");
+    getCreateTableDDLStatement();
+    LOG.info("Index creation completed.");
   }
 
-  @Override
-  public String getName() {
-    return "gcp-spanner";
+  private void getCreateTableDDLStatement() throws IOException {
+    List<String> ddlStatements = new ArrayList<>();
+    ddlStatements.add(getCreateMetadataTableDDLStatement());
+    ddlStatements.add(getCreateMetadataPropsTableDDLStatement());
+    ddlStatements.addAll(getAllSearchIndexDDLStatements());
+    executeDDLStatements(ddlStatements);
+  }
+
+  /**
+   * <p>Key features of the schema:
+   *     <ul>
+   *         <li>**`metadata_id`:** An Unique which is used to identify the various
+   *         components of the pipeline metadata.
+   *         </li>
+   *         <li>**`namespace`:** It contains the namespace associated with a given entity.
+   *         </li>
+   *         <li>**`entity_type`:** It contains the type of entity.
+   *         </li>
+   *         <li>**`name`:** Name of the entity.
+   *         </li>
+   *         <li>**`created`:** creation time of the entity.</li>
+   *         <li>**`user`:** user scope related properties and tags</li>
+   *         <li>**`system`:**system scope related properties and tags </li>
+   *         <li>**`metadata_column`:** contains all the metadata related </li>
+   *         <li>user_tokens, system_tokens and text_tokens are the Tokenlists of User,
+   *         System on which we create search indexes and use them for user, system
+   *         and null scope searches respectively.</li>
+   *     </ul>
+   * </p>
+   * TODO(CDAP-21174): Add the TTl policies for cleanups.
+   */
+  private String getCreateMetadataTableDDLStatement() {
+    return String.format(
+      "CREATE TABLE IF NOT EXISTS %s (" + // metadata
+        "%s STRING(MAX) NOT NULL," +  // metadata_id
+        "%s STRING(MAX) NOT NULL," + // namespace
+        "%s STRING(MAX) NOT NULL," + // entity_type
+        "%s STRING(MAX) NOT NULL," + // name
+        "%s INT64," + // create_time
+        "%s STRING(MAX)," + // user
+        "%s STRING(MAX)," + // system
+        "%s JSON," + // metadata_column
+        "%s INT64 NOT NULL," + // version
+        "user_tokens TOKENLIST AS " + // user_tokens list
+        "(TOKENIZE_SUBSTRING(%s, support_relative_search=>TRUE)) HIDDEN," +
+        "system_tokens TOKENLIST AS " + // system_tokens list
+        "(TOKENIZE_SUBSTRING(%s, support_relative_search=>TRUE)) HIDDEN," +
+        "text_tokens TOKENLIST AS " + // text_tokens list
+        "(TOKENLIST_CONCAT([User_Tokens, System_Tokens])) HIDDEN," +
+        ") PRIMARY KEY (%s) ", // metadata_id
+      METADATA_TABLE,
+      MetadataTable.METADATA_ID_FIELD,
+      MetadataTable.NAMESPACE_FIELD,
+      MetadataTable.TYPE_FIELD,
+      MetadataTable.NAME_FIELD,
+      MetadataTable.CREATED_FIELD,
+      MetadataTable.USER_FIELD,
+      MetadataTable.SYSTEM_FIELD,
+      MetadataTable.METADATA_COLUMN_FIELD,
+      MetadataTable.VERSION,
+      MetadataTable.USER_FIELD,
+      MetadataTable.SYSTEM_FIELD,
+      MetadataTable.METADATA_ID_FIELD
+    );
+  }
+
+  /**
+   * <p>Key features of the schema:
+   *     <ul>
+   *         <li>**`metadata_id`:**An Unique which is used to identify the various
+   *               components of the pipeline metadata.
+   *         </li>
+   *         <li>**`namespace`:** It contains the namespace associated with a given entity.
+   *         </li>
+   *         <li>**`entity_type`:** It contains the type of entity.
+   *         </li>
+   *         <li>**`props_name`:** Contains the name of the properties/tags associated to an
+   *             entries in metadata table.
+   *         </li>
+   *         <li>**`props_scope`:** Contains the scope of the properties/tags associated to an
+   *             entries in metadata table.
+   *         </li>
+   *         <li>**`props_value`:** Contains the value of the properties/tags associated to an
+   *             entries in metadata table.
+   *         <li>value_tokens is the Tokenlist of Props_Value Column on which we create search
+   *         indexes and use of key:value type searches.
+   *         respectively.</li>
+   *     </ul>
+   * </p>
+   */
+  private String getCreateMetadataPropsTableDDLStatement() {
+    return String.format(
+      "CREATE TABLE IF NOT EXISTS %s (" +
+        "%s STRING(MAX) NOT NULL," +  // metadata_id
+        "%s STRING(MAX) NOT NULL," + // namespace
+        "%s STRING(MAX) NOT NULL," + // entity_type
+        "%s STRING(MAX) NOT NULL," + // name
+        "%s STRING(MAX)," + // scope
+        "%s STRING(MAX)," + // value
+        "value_tokens TOKENLIST AS " + // value_tokens list
+        "(TOKENIZE_SUBSTRING(%s, support_relative_search=>TRUE)) HIDDEN," +
+        ") PRIMARY KEY (%s, %s, %s) ," + // metadata_id, name, scope
+        "INTERLEAVE IN PARENT %s ON DELETE CASCADE",
+      METADATA_PROPS_TABLE,
+      MetadataPropsTable.METADATA_ID_FIELD,
+      MetadataPropsTable.NAMESPACE_FIELD,
+      MetadataPropsTable.TYPE_FIELD,
+      MetadataPropsTable.NESTED_NAME_FIELD,
+      MetadataPropsTable.NESTED_SCOPE_FIELD,
+      MetadataPropsTable.NESTED_VALUE_FIELD,
+      MetadataPropsTable.NESTED_VALUE_FIELD,
+      MetadataPropsTable.METADATA_ID_FIELD,
+      MetadataPropsTable.NESTED_NAME_FIELD,
+      MetadataPropsTable.NESTED_SCOPE_FIELD,
+      METADATA_TABLE
+    );
+  }
+
+  private List<String> getAllSearchIndexDDLStatements() {
+    List<String> ddlStatements = new ArrayList<>();
+
+    // Create SEARCH INDEX on the Tokenized column(user_tokens) of user column.
+    ddlStatements.add(String.format("CREATE SEARCH INDEX UserNgramIndex ON %s(user_tokens)", METADATA_TABLE));
+
+    // Creates SEARCH INDEX on the Tokenized column(system_tokens) of system column.
+    ddlStatements.add(String.format("CREATE SEARCH INDEX SystemNgramIndex ON %s(system_tokens)", METADATA_TABLE));
+
+    // Creates SEARCH INDEX on the Tokenized column(text_tokens) of user and system column.
+    ddlStatements.add(String.format("CREATE SEARCH INDEX TextNgramIndex ON %s(text_tokens)", METADATA_TABLE));
+
+    // Creates SEARCH INDEX on the Tokenized column(value_tokens) of props_value column.
+    ddlStatements.add(String.format("CREATE SEARCH INDEX ValueNgramIndex ON %s(value_tokens)", METADATA_PROPS_TABLE));
+
+    return ddlStatements;
+  }
+
+  /**
+   * Creates the necessary metadata tables and associated search indexes in the database.
+   * This method orchestrates the creation of the core metadata table, metadata_props table
+   * and search indexes to facilitate efficient querying of user,system, text, and value-based
+   * data within the metadata.
+   * TODO (CDAP-21176): Checking additional Schema updates for schema backward compatibility.
+   */
+  private void executeDDLStatements(List<String> ddlStatements) throws IOException {
+    if (ddlStatements.isEmpty()) {
+      LOG.debug("No ddl statements to execute");
+      return;
+    }
+    try {
+      Uninterruptibles.getUninterruptibly(
+        adminClient.updateDatabaseDdl(instanceId,
+                                      databaseId, ddlStatements, null));
+    } catch (ExecutionException e) {
+      Throwable cause = e.getCause();
+      if (cause instanceof SpannerException
+        && ((SpannerException) cause).getErrorCode() == ErrorCode.FAILED_PRECONDITION) {
+        LOG.debug("Concurrent statement execution error: ", e);
+      } else {
+        throw new IOException(e);
+      }
+    }
   }
 
   @Override
   public void dropIndex() throws IOException {
-    throw new IOException("NOT IMPLEMENTED");
+    List<String> statements = new ArrayList<>();
+    statements.add(String.format("DROP TABLE IF EXISTS %s", METADATA_TABLE));
+    statements.add(String.format("DROP TABLE IF EXISTS %s", METADATA_PROPS_TABLE));
+    executeDDLStatements(statements);
+    LOG.info("Metadata Tables dropped successfully.");
   }
 
   @Override
@@ -77,5 +296,13 @@ public Metadata read(Read read) throws IOException {
   public SearchResponse search(SearchRequest request) throws IOException {
     throw new IOException("NOT IMPLEMENTED");
   }
+
+  @Override
+  public synchronized void close() {
+    if (spanner != null) {
+      spanner.close();
+      spanner = null;
+    }
+  }
 }
 
