Skip to content

Commit 00dade1

Browse files
author
Komal Yadav
committed
Create_tables
1 parent 14ce5fc commit 00dade1

2 files changed

Lines changed: 234 additions & 36 deletions

File tree

‎cdap-master/pom.xml‎

Lines changed: 1 addition & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -113,11 +113,6 @@
113113
<artifactId>cdap-messaging-ext-spanner</artifactId>
114114
<version>${project.version}</version>
115115
</dependency>
116-
<dependency>
117-
<groupId>io.cdap.cdap</groupId>
118-
<artifactId>cdap-metadata-ext-spanner</artifactId>
119-
<version>${project.version}</version>
120-
</dependency>
121116
<dependency>
122117
<groupId>org.apache.tephra</groupId>
123118
<artifactId>tephra-api</artifactId>
@@ -256,7 +251,6 @@
256251
<stage.authenticators.ext.dir>${stage.opt.dir}/ext/authenticators</stage.authenticators.ext.dir>
257252
<stage.credential.provider.extensions.dir>${stage.opt.dir}/ext/credentialproviders</stage.credential.provider.extensions.dir>
258253
<stage.messaging.ext.dir>${stage.opt.dir}/ext/messagingproviders</stage.messaging.ext.dir>
259-
<stage.metadata.ext.dir>${stage.opt.dir}/ext/metadatastorage</stage.metadata.ext.dir>
260254
<stage.metricswriters.ext.dir>${stage.opt.dir}/ext/metricswriters/google_cloud_monitoring_writer
261255
</stage.metricswriters.ext.dir>
262256
<stage.eventwriters.ext.dir>${stage.opt.dir}/ext/eventwriters/google_cloud_pubsub_writer
@@ -739,28 +733,6 @@
739733
</configuration>
740734
</execution>
741735

742-
<!-- Copy metadata extensions -->
743-
<execution>
744-
<id>copy-metadata-ext-spanner</id>
745-
<phase>process-resources</phase>
746-
<goals>
747-
<goal>copy-resources</goal>
748-
</goals>
749-
<configuration combine.self="override">
750-
<outputDirectory>${stage.metadata.ext.dir}/spanner</outputDirectory>
751-
<resources>
752-
<resource>
753-
<directory>
754-
${project.parent.basedir}/cdap-metadata-ext-spanner/target/libexec/
755-
</directory>
756-
<includes>
757-
<include>*.jar</include>
758-
</includes>
759-
</resource>
760-
</resources>
761-
</configuration>
762-
</execution>
763-
764736
<!-- Copy remote authenticator extensions -->
765737
<execution>
766738
<id>copy-authenticator-ext-gcp</id>
@@ -848,7 +820,7 @@
848820
<exclude name="${additional.artifacts.exclude.pattern}"/>
849821
<exclude name="**/target/*-sources.jar"/>
850822
<exclude name="**/target/*-javadoc.jar"/>
851-
<!-- Wrangler excludes -->
823+
<!-- Wrangler excludes -->
852824
<exclude name="**/target/wrangler-core*"/>
853825
<exclude name="**/target/wrangler-test*"/>
854826
<!--We don't want to package the following plugins with CDAP-->

‎cdap-metadata-ext-spanner/src/main/java/io/cdap/cdap/metadata/spanner/SpannerMetadataStorage.java‎

Lines changed: 233 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,12 @@
1616

1717
package io.cdap.cdap.metadata.spanner;
1818

19+
import com.google.cloud.spanner.DatabaseAdminClient;
20+
import com.google.cloud.spanner.ErrorCode;
21+
import com.google.cloud.spanner.Spanner;
22+
import com.google.cloud.spanner.SpannerException;
23+
import com.google.cloud.spanner.SpannerOptions;
24+
import com.google.common.util.concurrent.Uninterruptibles;
1925
import io.cdap.cdap.spi.metadata.Metadata;
2026
import io.cdap.cdap.spi.metadata.MetadataChange;
2127
import io.cdap.cdap.spi.metadata.MetadataMutation;
@@ -25,36 +31,247 @@
2531
import io.cdap.cdap.spi.metadata.Read;
2632
import io.cdap.cdap.spi.metadata.SearchRequest;
2733
import io.cdap.cdap.spi.metadata.SearchResponse;
34+
import org.slf4j.Logger;
35+
import org.slf4j.LoggerFactory;
2836
import java.io.IOException;
37+
import java.util.ArrayList;
2938
import java.util.List;
39+
import java.util.Map;
40+
import java.util.Objects;
41+
import java.util.concurrent.ExecutionException;
3042

3143
/**
3244
* A metadata storage provider that delegates to Spanner.
3345
*/
3446
public class SpannerMetadataStorage implements MetadataStorage {
3547

48+
private static final Logger LOG = LoggerFactory.getLogger(SpannerMetadataStorage.class);
49+
50+
// Metadata table names
51+
private static final String METADATA_TABLE = "metadata";
52+
private static final String METADATA_PROPS_TABLE = "metadata_props";
53+
54+
// 'metadata' table fields
55+
private static final String METADATA_ID_FIELD = "metadata_id";
56+
private static final String METADATA_COLUMN_FIELD = "metadata_column";
57+
private static final String NAMESPACE_FIELD = "namespace";
58+
private static final String TYPE_FIELD = "entity_type";
59+
private static final String NAME_FIELD = "name";
60+
private static final String VERSION = "version";
61+
private static final String CREATED_FIELD = "creation_time";
62+
private static final String USER_FIELD = "user";
63+
private static final String SYSTEM_FIELD = "system";
64+
65+
// 'metadata_props' table fields
66+
private static final String NESTED_NAME_FIELD = "props_name";
67+
private static final String NESTED_SCOPE_FIELD = "props_scope";
68+
private static final String NESTED_VALUE_FIELD = "props_value";
69+
70+
private String instanceId = "instance";
71+
private String projectId = "project";
72+
private String databaseId = "database";
73+
74+
private Spanner spanner;
75+
private DatabaseAdminClient adminClient;
76+
3677
@Override
3778
public void initialize(MetadataStorageContext context) throws Exception {
38-
throw new IOException("NOT IMPLEMENTED");
79+
Map<String, String> cConf = context.getProperties();
80+
81+
projectId = Objects.requireNonNull(cConf.get(projectId));
82+
instanceId = Objects.requireNonNull(cConf.get(projectId));
83+
databaseId = Objects.requireNonNull(cConf.get(projectId));
84+
85+
SpannerOptions.Builder builder = SpannerOptions.newBuilder().setProjectId(projectId);
86+
SpannerOptions options = builder.build();
87+
88+
this.spanner = options.getService();
89+
this.adminClient = spanner.getDatabaseAdminClient();
90+
91+
LOG.info("SpannerMetadataStorage initialized.");
3992
}
4093

4194
@Override
42-
public void close() {
95+
public String getName() {
96+
return "gcp-spanner";
4397
}
4498

4599
@Override
46100
public void createIndex() throws IOException {
47-
throw new IOException("NOT IMPLEMENTED");
101+
getCreateTableDDLStatement();
102+
LOG.info("Index creation completed.");
48103
}
49104

50-
@Override
51-
public String getName() {
52-
return "gcp-spanner";
105+
private void getCreateTableDDLStatement() throws IOException {
106+
List<String> DDLStatements = new ArrayList<>();
107+
DDLStatements.add(getCreateMetadataTableDDLStatement());
108+
DDLStatements.add(getCreateMetadataPropsTableDDLStatement());
109+
DDLStatements.addAll(getAllSearchIndexDDLStatements());
110+
executeDDLStatements(DDLStatements);
111+
}
112+
113+
/**
114+
* <p>Key features of the schema:
115+
* <ul>
116+
* <li>**`metadata_id`:** An Unique which is used to identify the various
117+
* components of the pipeline metadata.
118+
* </li>
119+
* <li>**`namespace`:** It contains the namespace associated with a given entity.
120+
* </li>
121+
* <li>**`entity_type`:** It contains the type of entity.
122+
* </li>
123+
* <li>**`name`:** Name of the entity.
124+
* </li>
125+
* <li>**`created`:** creation time of the entity.</li>
126+
* <li>**`user`:** user scope related properties and tags</li>
127+
* <li>**`system`:**system scope related properties and tags </li>
128+
* <li>**`metadata_column`:** contains all the metadata related </li>
129+
* <li>user_tokens, system_tokens and text_tokens are the Tokenlists of User,
130+
* System on which we create search indexes and use them for user, system
131+
* and null scope searches respectively.</li>
132+
* </ul>
133+
* </p>
134+
* TODO(CDAP-21174): Add the TTl policies for cleanups.
135+
*/
136+
private String getCreateMetadataTableDDLStatement() {
137+
return String.format(
138+
"CREATE TABLE IF NOT EXISTS %s (" + // metadata
139+
"%s STRING(MAX) NOT NULL," + // metadata_id
140+
"%s STRING(MAX) NOT NULL," + // namespace
141+
"%s STRING(MAX) NOT NULL," + // entity_type
142+
"%s STRING(MAX) NOT NULL," + // name
143+
"%s INT64," + // creation-time
144+
"%s STRING(MAX)," + // user
145+
"%s STRING(MAX)," + // system
146+
"%s JSON," + // metadata_column
147+
"%s INT64 NOT NULL," + // version
148+
"user_tokens TOKENLIST AS " + // user-tokens list
149+
"(TOKENIZE_SUBSTRING(%s, support_relative_search=>TRUE)) HIDDEN," +
150+
"system_tokens TOKENLIST AS " + // system-tokens list
151+
"(TOKENIZE_SUBSTRING(%s, support_relative_search=>TRUE)) HIDDEN," +
152+
"text_tokens TOKENLIST AS " + // text-tokens list
153+
"(TOKENLIST_CONCAT([User_Tokens, System_Tokens])) HIDDEN," +
154+
") PRIMARY KEY (%s) ", // metadata_id
155+
METADATA_TABLE,
156+
METADATA_ID_FIELD,
157+
NAMESPACE_FIELD,
158+
TYPE_FIELD,
159+
NAME_FIELD,
160+
CREATED_FIELD,
161+
USER_FIELD,
162+
SYSTEM_FIELD,
163+
METADATA_COLUMN_FIELD,
164+
VERSION,
165+
USER_FIELD,
166+
SYSTEM_FIELD,
167+
METADATA_ID_FIELD
168+
);
169+
}
170+
171+
/**
172+
* <p>Key features of the schema:
173+
* <ul>
174+
* <li>**`metadata_id`:**An Unique which is used to identify the various
175+
* components of the pipeline metadata.
176+
* </li>
177+
* <li>**`namespace`:** It contains the namespace associated with a given entity.
178+
* </li>
179+
* <li>**`entity_type`:** It contains the type of entity.
180+
* </li>
181+
* <li>**`props_name`:** Contains the name of the properties/tags associated to an
182+
* entries in metadata table.
183+
* </li>
184+
* <li>**`props_scope`:** Contains the scope of the properties/tags associated to an
185+
* entries in metadata table.
186+
* </li>
187+
* <li>**`props_value`:** Contains the value of the properties/tags associated to an
188+
* entries in metadata table.
189+
* <li>value_tokens is the Tokenlist of Props_Value Column on which we create search
190+
* indexes and use of key:value type searches.
191+
* respectively.</li>
192+
* </ul>
193+
* </p>
194+
*/
195+
private String getCreateMetadataPropsTableDDLStatement() {
196+
return String.format(
197+
"CREATE TABLE IF NOT EXISTS %s (" +
198+
"%s STRING(MAX) NOT NULL," + // metadata_id
199+
"%s STRING(MAX) NOT NULL," + // namespace
200+
"%s STRING(MAX) NOT NULL," + // entity_type
201+
"%s STRING(MAX) NOT NULL," + // name
202+
"%s STRING(MAX)," + // scope
203+
"%s STRING(MAX)," + // value
204+
"value_tokens TOKENLIST AS " + // value-tokens list
205+
"(TOKENIZE_SUBSTRING(%s, support_relative_search=>TRUE)) HIDDEN," +
206+
") PRIMARY KEY (%s, %s, %s) ," + // metadata_id, name, scope
207+
"INTERLEAVE IN PARENT %s ON DELETE CASCADE",
208+
METADATA_PROPS_TABLE,
209+
METADATA_ID_FIELD,
210+
NAMESPACE_FIELD,
211+
TYPE_FIELD,
212+
NESTED_NAME_FIELD,
213+
NESTED_SCOPE_FIELD,
214+
NESTED_VALUE_FIELD,
215+
NESTED_VALUE_FIELD,
216+
METADATA_ID_FIELD,
217+
NESTED_NAME_FIELD,
218+
NESTED_SCOPE_FIELD,
219+
METADATA_TABLE
220+
);
221+
}
222+
223+
private List<String> getAllSearchIndexDDLStatements() {
224+
List<String> DDLStatements = new ArrayList<>();
225+
226+
// Create SEARCH INDEX on the Tokenized column(user_tokens) of user column.
227+
DDLStatements.add(String.format("CREATE SEARCH INDEX UserNgramIndex ON %s(user_tokens)", METADATA_TABLE));
228+
229+
// Creates SEARCH INDEX on the Tokenized column(system_tokens) of system column.
230+
DDLStatements.add(String.format("CREATE SEARCH INDEX SystemNgramIndex ON %s(system_tokens)", METADATA_TABLE));
231+
232+
// Creates SEARCH INDEX on the Tokenized column(text_tokens) of user and system column.
233+
DDLStatements.add(String.format("CREATE SEARCH INDEX TextNgramIndex ON %s(text_tokens)", METADATA_TABLE));
234+
235+
// Creates SEARCH INDEX on the Tokenized column(value_tokens) of props_value column.
236+
DDLStatements.add(String.format("CREATE SEARCH INDEX ValueNgramIndex ON %s(value_tokens)", METADATA_PROPS_TABLE));
237+
238+
return DDLStatements;
239+
}
240+
241+
/**
242+
* Creates the necessary metadata tables and associated search indexes in the database.
243+
* This method orchestrates the creation of the core metadata table, metadata_props table
244+
* and search indexes to facilitate efficient querying of user,system, text, and value-based
245+
* data within the metadata.
246+
* TODO (CDAP-21176): Checking additional Schema updates for schema backward compatibility.
247+
*/
248+
private void executeDDLStatements(List<String> DDLStatements) throws IOException {
249+
if (DDLStatements.isEmpty()) {
250+
LOG.debug("No DDL statements to execute");
251+
return;
252+
}
253+
try {
254+
Uninterruptibles.getUninterruptibly(
255+
adminClient.updateDatabaseDdl(instanceId,
256+
databaseId, DDLStatements, null));
257+
} catch (ExecutionException e) {
258+
Throwable cause = e.getCause();
259+
if (cause instanceof SpannerException
260+
&& ((SpannerException) cause).getErrorCode() == ErrorCode.FAILED_PRECONDITION) {
261+
LOG.debug("Concurrent statement execution error: ", e);
262+
} else {
263+
throw new IOException(e);
264+
}
265+
}
53266
}
54267

55268
@Override
56269
public void dropIndex() throws IOException {
57-
throw new IOException("NOT IMPLEMENTED");
270+
List<String> statements = new ArrayList<>();
271+
statements.add(String.format("DROP TABLE IF EXISTS %s", METADATA_TABLE));
272+
statements.add(String.format("DROP TABLE IF EXISTS %s", METADATA_PROPS_TABLE));
273+
executeDDLStatements(statements);
274+
LOG.info("Metadata Tables dropped successfully.");
58275
}
59276

60277
@Override
@@ -77,5 +294,14 @@ public Metadata read(Read read) throws IOException {
77294
public SearchResponse search(SearchRequest request) throws IOException {
78295
throw new IOException("NOT IMPLEMENTED");
79296
}
297+
298+
@Override
299+
public synchronized void close() {
300+
if (spanner == null) {
301+
return;
302+
}
303+
spanner.close();
304+
spanner = null;
305+
}
80306
}
81307

0 commit comments

Comments
 (0)