Skip to content

Commit 779eaee

Browse files
jywadhwanijywadhwani
authored
metadata-models 72.0.8 -> 80.0.0 (#1756)
1 parent c5738d6 commit 779eaee

File tree

37 files changed

+861
-80
lines changed

37 files changed

+861
-80
lines changed

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@
1111
# Pegasus & Avro
1212
**/src/mainGenerated*
1313
**/src/testGenerated*
14+
metadata-events/mxe-registration/src/main/resources/**/*.avsc
1415

1516
# Java
1617
.java-version

‎metadata-dao-impl/ebean-dao/gma-create-all.sql‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,3 +15,18 @@ create table metadata_aspect (
1515
constraint pk_metadata_aspect primary key (urn,aspect,version)
1616
);
1717

18+
create table metadata_index (
19+
id bigint auto_increment not null,
20+
urn varchar(500) not null,
21+
aspect varchar(200) not null,
22+
path varchar(200) not null,
23+
longval bigint,
24+
stringval varchar(500),
25+
doubleval double,
26+
constraint pk_metadata_index primary key (id)
27+
);
28+
29+
create index idx_long_val on metadata_index (aspect,path,longval,urn);
30+
create index idx_string_val on metadata_index (aspect,path,stringval,urn);
31+
create index idx_double_val on metadata_index (aspect,path,doubleval,urn);
32+
create index idx_urn on metadata_index (urn);

‎metadata-dao-impl/ebean-dao/gma-drop-all.sql‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,3 +2,9 @@ drop table if exists metadata_id;
22

33
drop table if exists metadata_aspect;
44

5+
drop table if exists metadata_index;
6+
7+
drop index if exists idx_long_val;
8+
drop index if exists idx_string_val;
9+
drop index if exists idx_double_val;
10+
drop index if exists idx_urn;

‎metadata-dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/EbeanMetadataIndex.java‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
package com.linkedin.metadata.dao;
2+
3+
import io.ebean.annotation.Index;
4+
import io.ebean.Model;
5+
import javax.persistence.Column;
6+
import javax.persistence.Entity;
7+
import javax.persistence.GeneratedValue;
8+
import javax.persistence.GenerationType;
9+
import javax.persistence.Id;
10+
import javax.persistence.Table;
11+
import lombok.Getter;
12+
import lombok.NonNull;
13+
import lombok.Setter;
14+
15+
16+
@Getter
17+
@Setter
18+
// define composite indexes
19+
@Index(name = "idx_long_val", columnNames = {
20+
EbeanMetadataIndex.ASPECT_COLUMN,
21+
EbeanMetadataIndex.PATH_COLUMN,
22+
EbeanMetadataIndex.LONG_COLUMN,
23+
EbeanMetadataIndex.URN_COLUMN
24+
})
25+
@Index(name = "idx_string_val", columnNames = {
26+
EbeanMetadataIndex.ASPECT_COLUMN,
27+
EbeanMetadataIndex.PATH_COLUMN,
28+
EbeanMetadataIndex.STRING_COLUMN,
29+
EbeanMetadataIndex.URN_COLUMN
30+
})
31+
@Index(name = "idx_double_val", columnNames = {
32+
EbeanMetadataIndex.ASPECT_COLUMN,
33+
EbeanMetadataIndex.PATH_COLUMN,
34+
EbeanMetadataIndex.DOUBLE_COLUMN,
35+
EbeanMetadataIndex.URN_COLUMN
36+
})
37+
@Entity
38+
@Table(name = "metadata_index")
39+
public class EbeanMetadataIndex extends Model {
40+
41+
public static final long serialVersionUID = 1L;
42+
43+
private static final String ID_COLUMN = "id";
44+
public static final String URN_COLUMN = "urn";
45+
public static final String ASPECT_COLUMN = "aspect";
46+
public static final String PATH_COLUMN = "path";
47+
public static final String LONG_COLUMN = "longVal";
48+
public static final String STRING_COLUMN = "stringVal";
49+
public static final String DOUBLE_COLUMN = "doubleVal";
50+
51+
@Id
52+
@GeneratedValue(strategy = GenerationType.IDENTITY)
53+
@Column(name = ID_COLUMN)
54+
protected long id;
55+
56+
@NonNull
57+
@Index(name = "idx_urn")
58+
@Column(name = URN_COLUMN, length = 500, nullable = false)
59+
protected String urn;
60+
61+
@NonNull
62+
@Column(name = ASPECT_COLUMN, length = 200, nullable = false)
63+
protected String aspect;
64+
65+
@NonNull
66+
@Column(name = PATH_COLUMN, length = 200, nullable = false)
67+
protected String path;
68+
69+
@Column(name = LONG_COLUMN)
70+
protected Long longVal;
71+
72+
@Column(name = STRING_COLUMN, length = 500)
73+
protected String stringVal;
74+
75+
@Column(name = DOUBLE_COLUMN)
76+
protected Double doubleVal;
77+
78+
}

‎metadata-dao-impl/ebean-dao/src/main/resources/gma-create-all.sql‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,3 +15,18 @@ create table metadata_id (
1515
constraint uq_metadata_id_namespace_id unique (namespace,id)
1616
);
1717

18+
create table metadata_index (
19+
id bigint auto_increment not null,
20+
urn varchar(500) not null,
21+
aspect varchar(200) not null,
22+
path varchar(200) not null,
23+
longval bigint,
24+
stringval varchar(500),
25+
doubleval double,
26+
constraint pk_metadata_index primary key (id)
27+
);
28+
29+
create index idx_long_val on metadata_index (aspect,path,longval,urn);
30+
create index idx_string_val on metadata_index (aspect,path,stringval,urn);
31+
create index idx_double_val on metadata_index (aspect,path,doubleval,urn);
32+
create index idx_urn on metadata_index (urn);

‎metadata-dao-impl/elasticsearch-dao/src/main/java/com/linkedin/metadata/dao/utils/ESUtils.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,10 @@ public class ESUtils {
1818
private static final String DEFAULT_SEARCH_RESULTS_SORT_BY_FIELD = "urn";
1919

2020
/*
21-
* TODO: we might need to extend this list if need be, below link has the complete list
22-
* https://www.elastic.co/guide/en/elasticsearch/reference/current/regexp-syntax.html
23-
* */
24-
private static final char[] ELASTICSEARCH_REGEXP_RESERVED_CHARACTERS = {'*'};
21+
* Refer to https://www.elastic.co/guide/en/elasticsearch/reference/current/regexp-syntax.html for list of reserved
22+
* characters in an Elasticsearch regular expression.
23+
*/
24+
private static final String ELASTICSEARCH_REGEXP_RESERVED_CHARACTERS = "?+*|{}[]()";
2525

2626
private ESUtils() {
2727

@@ -81,7 +81,7 @@ public static void buildSortOrder(@Nonnull SearchSourceBuilder searchSourceBuild
8181
*/
8282
@Nonnull
8383
public static String escapeReservedCharacters(@Nonnull String input) {
84-
for (char reservedChar : ELASTICSEARCH_REGEXP_RESERVED_CHARACTERS) {
84+
for (char reservedChar : ELASTICSEARCH_REGEXP_RESERVED_CHARACTERS.toCharArray()) {
8585
input = input.replace(String.valueOf(reservedChar), "\\" + reservedChar);
8686
}
8787
return input;

‎metadata-dao-impl/elasticsearch-dao/src/test/java/com/linkedin/metadata/dao/ESUtilsTest.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,5 +39,7 @@ public void testBuildFilterQuery() throws Exception {
3939
public void testEscapeReservedCharacters() {
4040
assertEquals(escapeReservedCharacters("foobar"), "foobar");
4141
assertEquals(escapeReservedCharacters("**"), "\\*\\*");
42+
assertEquals(escapeReservedCharacters("()"), "\\(\\)");
43+
assertEquals(escapeReservedCharacters("{}"), "\\{\\}");
4244
}
4345
}

‎metadata-dao-impl/kafka-producer/src/main/java/com/linkedin/metadata/dao/producer/KafkaMetadataEventProducer.java‎

Lines changed: 51 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,17 +8,21 @@
88
import com.linkedin.metadata.dao.utils.ModelUtils;
99
import com.linkedin.metadata.dao.utils.RecordUtils;
1010
import com.linkedin.metadata.snapshot.Snapshot;
11+
import com.linkedin.mxe.Configs;
1112
import com.linkedin.mxe.MetadataAuditEvent;
1213
import com.linkedin.mxe.MetadataChangeEvent;
1314
import com.linkedin.mxe.Topics;
1415
import java.io.IOException;
16+
import java.lang.reflect.InvocationTargetException;
17+
import java.util.Arrays;
1518
import java.util.Collections;
1619
import java.util.List;
1720
import java.util.Optional;
1821
import javax.annotation.Nonnull;
1922
import javax.annotation.Nullable;
2023
import org.apache.avro.generic.GenericRecord;
2124
import org.apache.avro.generic.IndexedRecord;
25+
import org.apache.avro.specific.SpecificRecord;
2226
import org.apache.kafka.clients.producer.Callback;
2327
import org.apache.kafka.clients.producer.Producer;
2428
import org.apache.kafka.clients.producer.ProducerRecord;
@@ -107,11 +111,57 @@ record = EventUtils.pegasusToAvroMAE(metadataAuditEvent);
107111
}
108112
}
109113

114+
@Override
115+
public <ASPECT extends RecordTemplate> void produceAspectSpecificMetadataAuditEvent(@Nonnull URN urn,
116+
@Nullable ASPECT oldValue, @Nonnull ASPECT newValue) {
117+
118+
validateAspectSpecificTopic(ModelUtils.getAspectSpecificMAETopicName(urn, newValue));
119+
120+
String topic;
121+
Class<? extends SpecificRecord> maeAvroClass;
122+
RecordTemplate metadataAuditEvent;
123+
try {
124+
topic = (String) Topics.class.getField(ModelUtils.getAspectSpecificMAETopicName(urn, newValue)).get(null);
125+
maeAvroClass = Configs.TOPIC_SCHEMA_CLASS_MAP.get(topic);
126+
metadataAuditEvent = (RecordTemplate) EventUtils.getPegasusClass(maeAvroClass).newInstance();
127+
128+
metadataAuditEvent.getClass().getMethod("setUrn", urn.getClass()).invoke(metadataAuditEvent, urn);
129+
metadataAuditEvent.getClass().getMethod("setNewValue", newValue.getClass()).invoke(metadataAuditEvent, newValue);
130+
if (oldValue != null) {
131+
metadataAuditEvent.getClass()
132+
.getMethod("setOldValue", oldValue.getClass())
133+
.invoke(metadataAuditEvent, oldValue);
134+
}
135+
} catch (NoSuchFieldException | IllegalAccessException | ClassNotFoundException | NoSuchMethodException
136+
| InstantiationException | InvocationTargetException e) {
137+
throw new IllegalArgumentException("Failed to compose the Pegasus aspect specific MAE", e);
138+
}
139+
140+
GenericRecord record;
141+
try {
142+
record = EventUtils.pegasusToAvroAspectSpecificMXE(maeAvroClass, metadataAuditEvent);
143+
} catch (NoSuchFieldException | IOException | IllegalAccessException e) {
144+
throw new ModelConversionException("Failed to convert Pegasus aspect specific MAE to Avro", e);
145+
}
146+
147+
if (_callback.isPresent()) {
148+
_producer.send(new ProducerRecord(topic, urn.toString(), record), _callback.get());
149+
} else {
150+
_producer.send(new ProducerRecord(topic, urn.toString(), record));
151+
}
152+
}
153+
110154
@Nonnull
111155
private Snapshot makeSnapshot(@Nonnull URN urn, @Nonnull RecordTemplate value) {
112156
Snapshot snapshot = new Snapshot();
113157
List<ASPECT_UNION> aspects = Collections.singletonList(ModelUtils.newAspectUnion(_aspectUnionClass, value));
114158
RecordUtils.setSelectedRecordTemplateInUnion(snapshot, ModelUtils.newSnapshot(_snapshotClass, urn, aspects));
115159
return snapshot;
116160
}
117-
}
161+
162+
static void validateAspectSpecificTopic(@Nonnull String topic) {
163+
if (!Arrays.stream(Topics.class.getFields()).anyMatch(field -> field.getName().equals(topic))) {
164+
throw new IllegalArgumentException(String.format("The aspect specific topic %s is not registered.", topic));
165+
}
166+
}
167+
}

‎metadata-dao-impl/neo4j-dao/src/main/java/com/linkedin/metadata/dao/Neo4jQueryDAO.java‎

Lines changed: 17 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,11 @@
77
import com.linkedin.metadata.query.RelationshipFilter;
88
import com.linkedin.metadata.validator.EntityValidator;
99
import com.linkedin.metadata.validator.RelationshipValidator;
10+
import java.util.ArrayList;
1011
import java.util.HashMap;
1112
import java.util.List;
1213
import java.util.Map;
1314
import java.util.function.Function;
14-
import java.util.stream.Collectors;
1515
import java.util.stream.StreamSupport;
1616
import javax.annotation.Nonnull;
1717
import javax.annotation.Nullable;
@@ -185,7 +185,7 @@ public List<RecordTemplate> findMixedTypesRelationships(@Nonnull Statement query
185185

186186
@Nonnull
187187
public <SRC_ENTITY extends RecordTemplate, DEST_ENTITY extends RecordTemplate, RELATIONSHIP extends RecordTemplate>
188-
List<List<RecordTemplate>> getPathsToAllNodesTraversed(
188+
List<List<RecordTemplate>> getTraversedPaths(
189189
@Nullable Class<SRC_ENTITY> sourceEntityClass, @Nonnull Filter sourceEntityFilter,
190190
@Nullable Class<DEST_ENTITY> destinationEntityClass, @Nonnull Filter destinationEntityFilter,
191191
@Nonnull Class<RELATIONSHIP> relationshipType, @Nonnull RelationshipFilter relationshipFilter,
@@ -221,7 +221,7 @@ List<List<RecordTemplate>> getPathsToAllNodesTraversed(
221221

222222
final Statement statement = buildStatement(statementString, "length(p), dest.urn", offset, count);
223223

224-
return runQuery(statement, this::pathRecordToEntityList);
224+
return runQuery(statement, this::pathRecordToPathList);
225225
}
226226

227227
/**
@@ -302,11 +302,21 @@ private <ENTITY extends RecordTemplate> ENTITY nodeRecordToEntity(@Nonnull Class
302302
}
303303

304304
@Nonnull
305-
private List<RecordTemplate> pathRecordToEntityList(@Nonnull Record pathRecord) {
305+
private List<RecordTemplate> pathRecordToPathList(@Nonnull Record pathRecord) {
306306
final Path path = pathRecord.values().get(0).asPath();
307-
return StreamSupport.stream(path.nodes().spliterator(), false)
308-
.map(Neo4jUtil::nodeToEntity)
309-
.collect(Collectors.toList());
307+
final List<RecordTemplate> pathList = new ArrayList<>();
308+
309+
StreamSupport.stream(path.spliterator(), false)
310+
.map(Neo4jUtil::pathSegmentToRecordList)
311+
.forEach(segment -> {
312+
if (pathList.isEmpty()) {
313+
pathList.add(segment.get(0));
314+
}
315+
pathList.add(segment.get(1));
316+
pathList.add(segment.get(2));
317+
});
318+
319+
return pathList;
310320
}
311321

312322
@Nonnull

‎metadata-dao-impl/neo4j-dao/src/main/java/com/linkedin/metadata/dao/Neo4jUtil.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,14 +8,17 @@
88
import com.linkedin.metadata.query.Filter;
99
import com.linkedin.metadata.query.RelationshipDirection;
1010
import com.linkedin.metadata.query.RelationshipFilter;
11+
import java.util.Arrays;
1112
import java.util.Collections;
1213
import java.util.HashMap;
14+
import java.util.List;
1315
import java.util.Map;
1416
import java.util.StringJoiner;
1517
import javax.annotation.Nonnull;
1618
import javax.annotation.Nullable;
1719
import org.apache.commons.lang3.ClassUtils;
1820
import org.neo4j.driver.types.Node;
21+
import org.neo4j.driver.types.Path;
1922
import org.neo4j.driver.types.Relationship;
2023

2124
import static com.linkedin.metadata.dao.utils.QueryUtils.*;
@@ -165,6 +168,25 @@ public static RecordTemplate nodeToEntity(@Nonnull Node node) {
165168
return RecordUtils.toRecordTemplate(className, new DataMap(node.asMap()));
166169
}
167170

171+
/**
172+
* Converts path segment (field:value map) list of {@link RecordTemplate}s of nodes & edges
173+
*
174+
* @param segment The segment of a path containing nodes & edges
175+
* @return List<RecordTemplate>
176+
*/
177+
@Nonnull
178+
public static List<RecordTemplate> pathSegmentToRecordList(@Nonnull Path.Segment segment) {
179+
final Node startNode = segment.start();
180+
final Node endNode = segment.end();
181+
final Relationship edge = segment.relationship();
182+
183+
return Arrays.asList(
184+
nodeToEntity(startNode),
185+
edgeToRelationship(startNode, endNode, edge),
186+
nodeToEntity(endNode)
187+
);
188+
}
189+
168190
/**
169191
* Converts edge (source-relationship->destination) to RELATIONSHIP
170192
*

0 commit comments

Comments
 (0)