-/*******************************************************************************\r
- * Copyright (c) 2012, 2016 Association for Decentralized Information Management\r
- * in Industry THTH ry.\r
- * All rights reserved. This program and the accompanying materials\r
- * are made available under the terms of the Eclipse Public License v1.0\r
- * which accompanies this distribution, and is available at\r
- * http://www.eclipse.org/legal/epl-v10.html\r
- *\r
- * Contributors:\r
- * VTT Technical Research Centre of Finland - initial API and implementation\r
- * Semantum Oy\r
- *******************************************************************************/\r
-package org.simantics.graph.db;\r
-\r
-import java.io.DataInput;\r
-import java.io.DataOutput;\r
-import java.io.DataOutputStream;\r
-import java.io.FileNotFoundException;\r
-import java.io.FileOutputStream;\r
-import java.io.IOException;\r
-import java.io.InputStream;\r
-import java.util.HashSet;\r
-import java.util.Map;\r
-import java.util.Set;\r
-import java.util.TreeMap;\r
-\r
-import org.simantics.databoard.Bindings;\r
-import org.simantics.databoard.adapter.AdaptException;\r
-import org.simantics.databoard.binding.Binding;\r
-import org.simantics.databoard.binding.mutable.Variant;\r
-import org.simantics.databoard.serialization.Serializer;\r
-import org.simantics.databoard.type.Datatype;\r
-import org.simantics.db.ReadGraph;\r
-import org.simantics.db.Resource;\r
-import org.simantics.db.Session;\r
-import org.simantics.db.VirtualGraph;\r
-import org.simantics.db.WriteOnlyGraph;\r
-import org.simantics.db.common.WriteBindings;\r
-import org.simantics.db.common.procedure.adapter.TransientCacheAsyncListener;\r
-import org.simantics.db.common.request.PossibleIndexRoot;\r
-import org.simantics.db.common.uri.UnescapedChildMapOfResource;\r
-import org.simantics.db.common.utils.Logger;\r
-import org.simantics.db.exception.DatabaseException;\r
-import org.simantics.db.service.ClusterBuilder2;\r
-import org.simantics.db.service.ClusterBuilderFactory;\r
-import org.simantics.db.service.ClusteringSupport;\r
-import org.simantics.db.service.SerialisationSupport;\r
-import org.simantics.graph.db.TransferableGraphSource.TransferableGraphSourceProcedure;\r
-import org.simantics.graph.db.TransferableGraphSource.TransferableGraphSourceValueProcedure;\r
-import org.simantics.graph.representation.Extensions;\r
-import org.simantics.graph.representation.External;\r
-import org.simantics.graph.representation.Identity;\r
-import org.simantics.graph.representation.IdentityDefinition;\r
-import org.simantics.graph.representation.Internal;\r
-import org.simantics.graph.representation.Optional;\r
-import org.simantics.graph.representation.Root;\r
-import org.simantics.graph.representation.TransferableGraphUtils;\r
-import org.simantics.graph.utils.TGResourceUtil;\r
-import org.simantics.graph.utils.TGResourceUtil.LongAdapter;\r
-\r
-public class StreamingTransferableGraphImportProcess implements TransferableGraphImporter {\r
- \r
- public static String LOG_FILE = "transferableGraphs.log";\r
- final static private boolean LOG = false;\r
- \r
- static DataOutput log;\r
-\r
- static {\r
-\r
- if (LOG) {\r
- \r
- try {\r
- FileOutputStream stream = new FileOutputStream(LOG_FILE);\r
- log = new DataOutputStream(stream);\r
- } catch (FileNotFoundException e) {\r
- e.printStackTrace();\r
- }\r
- }\r
-\r
- }\r
- \r
- private static void log(String line) {\r
- if (LOG) {\r
- try {\r
- log.writeUTF(line + "\n");\r
- } catch (IOException e) {\r
- e.printStackTrace();\r
- }\r
- }\r
- }\r
-\r
- Resource indexRoot;\r
- TransferableGraphSource tg;\r
- VirtualGraph vg;\r
- IImportAdvisor2 advisor;\r
- ClusterBuilder2 builder;\r
- final TGResourceUtil resourceUtil = new TGResourceUtil();\r
- \r
- int[] handles;\r
- \r
- Set<String> missingExternals = new HashSet<String>(); \r
- \r
- int resourceCount;\r
- Identity[] identities;\r
- TreeMap<String, Variant> extensions;\r
- \r
- // Builtins\r
- Resource RootLibrary;\r
- Resource String;\r
- Resource Library;\r
- \r
- Resource InstanceOf;\r
- Resource ConsistsOf;\r
- Resource PartOf;\r
- Resource HasName;\r
- Resource NameOf; \r
- \r
- public StreamingTransferableGraphImportProcess(Session session, VirtualGraph vg, TransferableGraphSource tg, IImportAdvisor2 advisor) {\r
- this.tg = tg;\r
- this.vg = vg;\r
- this.advisor = advisor;\r
- }\r
- \r
- public void readIdentities(ReadGraph g) throws Exception {\r
- extensions = tg.getExtensions();\r
- resourceCount = tg.getResourceCount();\r
- identities = new Identity[tg.getIdentityCount()];\r
- tg.forIdentities(g, new TransferableGraphSourceProcedure<Identity>() {\r
- \r
- int counter = 0;\r
- \r
- @Override\r
- public void execute(Identity value) throws Exception {\r
- identities[counter++] = value;\r
- }\r
- });\r
- }\r
- \r
- public void findBuiltins(WriteOnlyGraph g) throws DatabaseException {\r
- RootLibrary = g.getBuiltin("http:/");\r
- String = g.getBuiltin(CoreInitialization.LAYER0 + "String");\r
- Library = g.getBuiltin(CoreInitialization.LAYER0 + "Library");\r
- InstanceOf = g.getBuiltin(CoreInitialization.LAYER0 + "InstanceOf");\r
- ConsistsOf = g.getBuiltin(CoreInitialization.LAYER0 + "ConsistsOf");\r
- PartOf = g.getBuiltin(CoreInitialization.LAYER0 + "PartOf");\r
- HasName = g.getBuiltin(CoreInitialization.LAYER0 + "HasName");\r
- NameOf = g.getBuiltin(CoreInitialization.LAYER0 + "NameOf");\r
- }\r
- \r
- public void findBuiltins(ReadGraph g) throws DatabaseException {\r
- RootLibrary = g.getBuiltin("http:/");\r
- String = g.getBuiltin(CoreInitialization.LAYER0 + "String");\r
- Library = g.getBuiltin(CoreInitialization.LAYER0 + "Library");\r
- InstanceOf = g.getBuiltin(CoreInitialization.LAYER0 + "InstanceOf");\r
- ConsistsOf = g.getBuiltin(CoreInitialization.LAYER0 + "ConsistsOf");\r
- PartOf = g.getBuiltin(CoreInitialization.LAYER0 + "PartOf");\r
- HasName = g.getBuiltin(CoreInitialization.LAYER0 + "HasName");\r
- NameOf = g.getBuiltin(CoreInitialization.LAYER0 + "NameOf");\r
- }\r
-\r
-// /* Preparation that is used when the core is empty. \r
-// */\r
-// void initialPrepare(WriteOnlyGraph graph) throws DatabaseException {\r
-// findBuiltins(graph);\r
-// \r
-// resources = new Resource[tg.resourceCount];\r
-// \r
-// int Root = -1;\r
-// int SimanticsDomain = -1;\r
-// int Layer0 = -1;\r
-// \r
-// for(Identity identity : tg.identities) {\r
-// if(identity.definition instanceof Internal) {\r
-// Internal def = (Internal)identity.definition;\r
-// Resource res = null;\r
-// if(def.parent == Layer0) {\r
-// try {\r
-// res = graph.getBuiltin(CoreInitialization.LAYER0 + def.name);\r
-// } catch(ResourceNotFoundException e) { \r
-// }\r
-// }\r
-// else if(def.parent == SimanticsDomain) {\r
-// if(def.name.equals("Layer0-1.0"))\r
-// Layer0 = identity.resource;\r
-// }\r
-// else if(def.parent == Root) {\r
-// if(def.name.equals("www.simantics.org"))\r
-// SimanticsDomain = identity.resource;\r
-// }\r
-//\r
-// if(res == null)\r
-// res = createChild(graph, resources[def.parent], def.name);\r
-// else\r
-// createChild(graph, res, resources[def.parent], def.name);\r
-// resources[identity.resource] = res;\r
-// }\r
-// else if(identity.definition instanceof Root) {\r
-// Root = identity.resource;\r
-// resources[identity.resource] = RootLibrary; \r
-// } \r
-// }\r
-// }\r
- \r
- void addMissing(String external) {\r
- Set<String> removals = new HashSet<String>();\r
- for(String ext : missingExternals) if(ext.startsWith(external)) return;\r
- for(String ext : missingExternals) if(external.startsWith(ext)) removals.add(ext);\r
- missingExternals.removeAll(removals);\r
- missingExternals.add(external);\r
- }\r
- \r
- void prepare(ReadGraph graph) throws Exception {\r
-\r
- Resource target = advisor.getTarget();\r
- if(target != null)\r
- indexRoot = graph.syncRequest(new PossibleIndexRoot(target));\r
- \r
- findBuiltins(graph);\r
- readIdentities(graph);\r
- \r
-// System.err.println("ext: " + extensions);\r
-// System.err.println("rc: " + resourceCount);\r
-// System.err.println("ic: " + identities.length);\r
- \r
- ClusterBuilderFactory factory = graph.getService(ClusterBuilderFactory.class);\r
- ClusterBuilder2 builder = factory.create(vg, false);\r
- \r
- this.handles = new int[resourceCount];\r
- \r
- for(Identity identity : identities) {\r
- IdentityDefinition definition = identity.definition;\r
- if(definition instanceof External) {\r
- External def = (External)definition;\r
- if(def.parent == -1) {\r
- handles[identity.resource] = builder.handle(RootLibrary);\r
- } else {\r
- if("@inverse".equals(def.name)) {\r
- int parent = handles[def.parent];\r
- int child = builder.handle(graph.getInverse(builder.resource(parent)));\r
- handles[identity.resource] = child;\r
- } else {\r
- int handle = handles[def.parent];\r
- Resource parent = handle != 0 ? builder.resource(handle) : null;\r
- // TODO: escape should be removed when names become well-behaving\r
- if(parent != null) {\r
- Map<String,Resource> childMap = graph\r
- .syncRequest(new UnescapedChildMapOfResource(parent),\r
- new TransientCacheAsyncListener<Map<String, Resource>>()); \r
- Resource child = childMap.get(def.name); \r
- if(child == null) {\r
- addMissing(graph.getURI(parent) + "/" + def.name);\r
- } else {\r
- handles[identity.resource] = builder.handle(child);\r
- }\r
- } else {\r
- addMissing(TransferableGraphUtils.getURI(resourceCount, identities, def.parent) + "/" + def.name);\r
- }\r
- }\r
- }\r
- }\r
- else if(definition instanceof Internal) {\r
- // Do not do anything for now\r
- }\r
- else if(definition instanceof Root) {\r
- Root root = (Root)definition;\r
- if(root.name.equals(""))\r
- handles[identity.resource] = builder.handle(RootLibrary);\r
- else {\r
- Resource existing = advisor.analyzeRoot(graph, root);\r
- if(existing != null)\r
- handles[identity.resource] = builder.handle(existing);\r
- }\r
- }\r
- else if(definition instanceof Optional) {\r
- External def = (External)definition;\r
- Resource parent = builder.resource(handles[def.parent]);\r
- if(parent != null)\r
- handles[identity.resource] = builder.handle(graph.syncRequest(new UnescapedChildMapOfResource(parent)).get(def.name)); \r
- }\r
- } \r
- \r
- if(!missingExternals.isEmpty()) throw new MissingDependencyException(this);\r
- \r
- }\r
-\r
- @Override\r
- public Resource createChild(WriteOnlyGraph graph, Resource parent, Resource child, String name) throws DatabaseException {\r
- if(child == null) child = graph.newResource();\r
- Resource nameResource = graph.newResource();\r
- graph.claim(nameResource, InstanceOf, null, String);\r
- graph.claimValue(nameResource, name, WriteBindings.STRING);\r
- graph.claim(child, HasName, NameOf, nameResource);\r
- return child;\r
- }\r
- \r
- int[] getClustering() {\r
- Variant v = extensions.get(Extensions.CLUSTERING);\r
- if(v == null) return null;\r
- try {\r
- return (int[])v.getValue(Bindings.INT_ARRAY);\r
- } catch (AdaptException e) {\r
- Logger.defaultLogError(e);\r
- return null;\r
- }\r
- }\r
-\r
- int[] getClusterSets() {\r
- Variant v = extensions.get(Extensions.CLUSTER_SETS);\r
- if(v == null) return null;\r
- try {\r
- return (int[])v.getValue(Bindings.INT_ARRAY);\r
- } catch (AdaptException e) {\r
- Logger.defaultLogError(e);\r
- return null;\r
- }\r
- }\r
-\r
- boolean needTranslation(Datatype type) {\r
- return resourceUtil.mayHaveResource(type);\r
- }\r
- \r
- void findClusterSet(WriteOnlyGraph graph, Resource rootLibrary, int[] clustering, int[] clusterSets, long[] clusters, int id) throws DatabaseException {\r
- ClusteringSupport support = graph.getService(ClusteringSupport.class);\r
- if(id == Extensions.ROOT_LIBRARY_CLUSTER_SET || id == Extensions.INDEX_ROOT_CLUSTER_SET) return;\r
- for(int pos=0,index=0;index<clustering.length;index++) {\r
- pos += clustering[index];\r
- if(id < pos) {\r
- int cs = clusterSets[index]; \r
- if(handles[id] == 0) {\r
- int csHandle = 0;\r
- if(cs == Extensions.ROOT_LIBRARY_CLUSTER_SET) csHandle = builder.handle(rootLibrary);\r
- else if(cs == Extensions.INDEX_ROOT_CLUSTER_SET) {\r
- if(indexRoot == null) throw new DatabaseException("No index root was available in TG import.");\r
- csHandle = builder.handle(indexRoot);\r
- }\r
- else {\r
- findClusterSet(graph, rootLibrary, clustering, clusterSets, clusters, cs);\r
- csHandle = handles[cs];\r
- }\r
- \r
- if(clusters[index] != 0)\r
- builder.selectCluster(clusters[index]);\r
- else if(cs >= 0)\r
- builder.newCluster(csHandle);\r
- \r
- handles[id] = builder.newResource(csHandle);\r
- clusters[index] = support.getCluster(builder.resource(handles[id]));\r
- \r
- builder.createClusterSet(handles[id]);\r
- }\r
- return;\r
- }\r
- }\r
- }\r
- \r
- void write(final WriteOnlyGraph graph) throws Exception {\r
- \r
- final SerialisationSupport ss = graph.getService(SerialisationSupport.class);\r
- \r
- ClusterBuilderFactory factory = graph.getService(ClusterBuilderFactory.class);\r
- if(advisor instanceof IImportAdvisor2) {\r
- boolean allowImmutable = ((IImportAdvisor2)advisor).allowImmutableModifications();\r
- builder = factory.create(vg, allowImmutable);\r
- } else {\r
- builder = factory.create(vg, false);\r
- }\r
- \r
- final int[] handles = this.handles; \r
- \r
- int[] clustering = getClustering();\r
- if(clustering != null) {\r
- \r
- int[] clusterSets = getClusterSets();\r
- if(clusterSets != null) {\r
-\r
- assert(clustering.length == clusterSets.length);\r
-\r
- long[] clusters = new long[clustering.length];\r
- \r
- // Create clustering\r
- for(int i=0;i<clusterSets.length;i++) {\r
- findClusterSet(graph, graph.getRootLibrary(), clustering, clusterSets, clusters, clusterSets[i]);\r
- }\r
- \r
- // Then create all resources\r
- int i=0;\r
- for(int j=0;j<clustering.length;j++) {\r
- int c = clustering[j];\r
- int s = clusterSets[j];\r
- int setHandle = 0;\r
- if(s == Extensions.ROOT_LIBRARY_CLUSTER_SET)\r
- setHandle = builder.handle(graph.getRootLibrary());\r
- else if(s == Extensions.INDEX_ROOT_CLUSTER_SET) {\r
- if(indexRoot == null) throw new DatabaseException("No index root was available in TG import.");\r
- setHandle = builder.handle(indexRoot);\r
- }\r
- else setHandle = handles[s];\r
- // Preserve clustering only for internal resources\r
- if(clusters[j] != 0)\r
- builder.selectCluster(clusters[j]);\r
- else if(s >= 0)\r
- builder.newCluster(setHandle);\r
- for(int r=0;r<c;r++, i++)\r
- if(handles[i] == 0) handles[i] = builder.newResource();\r
- }\r
-\r
- for(;i<handles.length;++i)\r
- if(handles[i] == 0) handles[i] = builder.newResource();\r
- \r
- } else {\r
-\r
- int i = 0;\r
- for(int c : clustering) {\r
- builder.newCluster();\r
- for(int r=0;r<c;r++, i++)\r
- if(handles[i] == 0) handles[i] = builder.newResource();\r
- }\r
-\r
- for(;i<handles.length;++i)\r
- if(handles[i] == 0) handles[i] = builder.newResource();\r
- \r
- }\r
- \r
- } else {\r
- \r
- // Create blank resources\r
- for(int i=0;i<handles.length;++i)\r
- if(handles[i] == 0) handles[i] = builder.newResource();\r
-\r
- }\r
- \r
- // Internal identities \r
- for(Identity identity : identities) {\r
- IdentityDefinition definition = identity.definition;\r
-// if(handles[identity.resource] != 0)\r
-// continue;\r
- if(definition instanceof External) {\r
- // Already done everything\r
- }\r
- else if(definition instanceof Internal) {\r
- Internal def = (Internal)definition;\r
- if(handles[identity.resource] != 0)\r
- handles[identity.resource] = builder.handle(advisor.createChild(graph, this, builder.resource(handles[def.parent]), builder.resource(handles[identity.resource]), def.name));\r
- else\r
- handles[identity.resource] = builder.handle(advisor.createChild(graph, this, builder.resource(handles[def.parent]), null, def.name));\r
- }\r
- else if(definition instanceof Root) {\r
- \r
- Root root = (Root)definition;\r
- if(handles[identity.resource] != 0)\r
- handles[identity.resource] = builder.handle(advisor.createRoot(graph, root, builder.resource(handles[identity.resource])));\r
- else\r
- handles[identity.resource] = builder.handle(advisor.createRoot(graph, root, null));\r
- }\r
- else if(definition instanceof Optional) {\r
- Optional def = (Optional)definition;\r
- if(handles[identity.resource] != 0) {\r
- Resource child = advisor.createChild(graph, this, builder.resource(handles[def.parent]), builder.resource(handles[identity.resource]), def.name);\r
- graph.claim(child, InstanceOf, null, Library); // ???\r
- handles[identity.resource] = builder.handle(child);\r
- } else {\r
- Resource child = advisor.createChild(graph, this, builder.resource(handles[def.parent]), null, def.name);\r
- graph.claim(child, InstanceOf, null, Library); // ???\r
- handles[identity.resource] = builder.handle(child);\r
- }\r
- }\r
- } \r
- \r
- tg.getStatementCount();\r
- tg.forStatements(null, new TransferableGraphSourceProcedure<int[]>() {\r
-\r
- @Override\r
- public void execute(int[] value) throws Exception {\r
- \r
- int sub = value[0];\r
- int pred = value[1];\r
- int inv = value[2];\r
- int obj = value[3];\r
-\r
- int subject = handles[sub];\r
- int predicate = handles[pred];\r
- int object = handles[obj];\r
-\r
- builder.addStatement(graph, subject, predicate, object); \r
- if(inv >= 0) {\r
- int inverse = handles[inv];\r
- builder.addStatement(graph, object, inverse, subject); \r
- }\r
- \r
- }\r
- \r
- }); \r
- \r
- tg.getValueCount();\r
-\r
- class ValueProcedure extends InputStream implements TransferableGraphSourceValueProcedure {\r
-\r
- private TGResourceUtil util = new TGResourceUtil();\r
- private DataInput source;\r
-\r
- @Override\r
- public void execute(int _resource, Datatype type, DataInput stream) throws Exception {\r
-\r
- source = stream;\r
-\r
- //int file = _resource & 0x80000000;\r
- int resource = _resource & 0x7FFFFFFF;\r
-\r
- Binding binding = Bindings.getBinding(type);\r
- Serializer s = Bindings.getSerializer(binding);\r
-\r
- builder.beginValue(handles[resource]);\r
- if(util.mayHaveResource(type)) {\r
- Object value = s.deserialize(stream);\r
- util.adaptValue( binding, value, new LongAdapter() {\r
- @Override\r
- public long adapt(long in) {\r
- try {\r
- return ss.getRandomAccessId(handles[(int)in]);\r
- } catch (DatabaseException e) {\r
- throw new IllegalStateException(e);\r
- }\r
- }\r
- });\r
- byte[] bytes = s.serialize(value);\r
- for(byte b : bytes) {\r
- int val = b;\r
- if(val < 0) val += 256;\r
- builder.appendValue(val);\r
- }\r
- } else {\r
- s.skip(this);\r
- }\r
- builder.endValue();\r
- \r
- }\r
-\r
- @Override\r
- public int read() throws IOException {\r
- int value = source.readUnsignedByte();\r
- try {\r
- builder.appendValue(value);\r
- } catch (DatabaseException e) {\r
- e.printStackTrace();\r
- }\r
- return value;\r
- }\r
-\r
- @Override\r
- public void rawCopy(int resource, int length, DataInput input) throws Exception {\r
- builder.beginValue(handles[resource]);\r
- for (int i = 0; i < length; ++i)\r
- builder.appendValue(input.readUnsignedByte());\r
- builder.endValue();\r
- }\r
-\r
- };\r
- \r
- tg.forValues2(null, new ValueProcedure());\r
- \r
- }\r
- \r
- @Override\r
- public long[] getResourceIds(SerialisationSupport serializer) throws DatabaseException {\r
- final int count = handles.length;\r
- long[] resourceIds = new long[count];\r
- for(int i=0;i<count;++i)\r
- resourceIds[i] = serializer.getRandomAccessId(handles[i]);\r
- return resourceIds;\r
- }\r
+/*******************************************************************************
+ * Copyright (c) 2012, 2017 Association for Decentralized Information Management
+ * in Industry THTH ry.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ * VTT Technical Research Centre of Finland - initial API and implementation
+ * Semantum Oy - e.g. #7016
+ *******************************************************************************/
+package org.simantics.graph.db;
+
+import java.io.DataInput;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+
+import org.simantics.databoard.Bindings;
+import org.simantics.databoard.adapter.AdaptException;
+import org.simantics.databoard.binding.Binding;
+import org.simantics.databoard.binding.mutable.Variant;
+import org.simantics.databoard.serialization.Serializer;
+import org.simantics.databoard.type.Datatype;
+import org.simantics.databoard.util.URIStringUtils;
+import org.simantics.db.ReadGraph;
+import org.simantics.db.Resource;
+import org.simantics.db.Session;
+import org.simantics.db.VirtualGraph;
+import org.simantics.db.WriteOnlyGraph;
+import org.simantics.db.common.WriteBindings;
+import org.simantics.db.common.procedure.adapter.TransientCacheAsyncListener;
+import org.simantics.db.common.request.PossibleIndexRoot;
+import org.simantics.db.common.uri.UnescapedChildMapOfResource;
+import org.simantics.db.common.utils.Logger;
+import org.simantics.db.exception.DatabaseException;
+import org.simantics.db.service.ClusterBuilder2;
+import org.simantics.db.service.ClusterBuilderFactory;
+import org.simantics.db.service.ClusteringSupport;
+import org.simantics.db.service.SerialisationSupport;
+import org.simantics.db.service.XSupport;
+import org.simantics.graph.db.TransferableGraphSource.TransferableGraphSourceProcedure;
+import org.simantics.graph.db.TransferableGraphSource.TransferableGraphSourceValueProcedure;
+import org.simantics.graph.representation.Extensions;
+import org.simantics.graph.representation.External;
+import org.simantics.graph.representation.Identity;
+import org.simantics.graph.representation.IdentityDefinition;
+import org.simantics.graph.representation.Internal;
+import org.simantics.graph.representation.Optional;
+import org.simantics.graph.representation.Root;
+import org.simantics.graph.representation.TransferableGraphUtils;
+import org.simantics.graph.utils.TGResourceUtil;
+import org.simantics.graph.utils.TGResourceUtil.LongAdapter;
+import org.simantics.utils.datastructures.Pair;
+import org.slf4j.LoggerFactory;
+
+import gnu.trove.map.TIntObjectMap;
+import gnu.trove.map.hash.TIntObjectHashMap;
+
+public class StreamingTransferableGraphImportProcess implements TransferableGraphImporter {
+
+ private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(StreamingTransferableGraphImportProcess.class);
+
+ Resource indexRoot;
+ TransferableGraphSource tg;
+ VirtualGraph vg;
+ IImportAdvisor2 advisor;
+ TGStatusMonitor monitor;
+ ClusterBuilder2 builder;
+ final TGResourceUtil resourceUtil = new TGResourceUtil();
+
+ int[] handles;
+
+ Map<String,Integer> allMissingExternals = new HashMap<>();
+ Set<String> missingExternals = new HashSet<>();
+ Map<String,Resource> resolvedParents = new HashMap<>();
+ TIntObjectHashMap<Resource> existingInternalMap = new TIntObjectHashMap<>();
+
+ int resourceCount;
+ Identity[] identities;
+ TreeMap<String, Variant> extensions;
+
+ // Builtins
+ Resource RootLibrary;
+ Resource String;
+ Resource ExternalEntity;
+ Resource Library;
+
+ Resource InstanceOf;
+ Resource ConsistsOf;
+ Resource PartOf;
+ Resource HasName;
+ Resource NameOf;
+
+ public StreamingTransferableGraphImportProcess(Session session, VirtualGraph vg, TransferableGraphSource tg, IImportAdvisor2 advisor) {
+ this(session, vg, tg, advisor, null);
+ }
+
+ public StreamingTransferableGraphImportProcess(Session session, VirtualGraph vg, TransferableGraphSource tg, IImportAdvisor2 advisor, TGStatusMonitor monitor) {
+ this.tg = tg;
+ this.vg = vg;
+ this.advisor = advisor;
+ this.monitor = monitor;
+ }
+
+ private int updatePercentage(int percentage, int done, int total) {
+ if (monitor != null && (done & 63) == 0) {
+ int current = 100*done / total;
+ if (current > percentage) {
+ percentage = current;
+ monitor.status(percentage);
+ }
+ }
+ return percentage;
+ }
+
+ public void readIdentities(ReadGraph g) throws Exception {
+ extensions = tg.getExtensions();
+ resourceCount = tg.getResourceCount();
+ identities = new Identity[tg.getIdentityCount()];
+ tg.forIdentities(g, new TransferableGraphSourceProcedure<Identity>() {
+
+ int counter = 0;
+
+ @Override
+ public void execute(Identity value) throws Exception {
+ identities[counter++] = value;
+ }
+ });
+ }
+
+ public void findBuiltins(WriteOnlyGraph g) throws DatabaseException {
+ RootLibrary = g.getBuiltin("http:/");
+ String = g.getBuiltin(CoreInitialization.LAYER0 + "String");
+ Library = g.getBuiltin(CoreInitialization.LAYER0 + "Library");
+ InstanceOf = g.getBuiltin(CoreInitialization.LAYER0 + "InstanceOf");
+ ConsistsOf = g.getBuiltin(CoreInitialization.LAYER0 + "ConsistsOf");
+ PartOf = g.getBuiltin(CoreInitialization.LAYER0 + "PartOf");
+ HasName = g.getBuiltin(CoreInitialization.LAYER0 + "HasName");
+ NameOf = g.getBuiltin(CoreInitialization.LAYER0 + "NameOf");
+ ExternalEntity = g.getBuiltin(CoreInitialization.LAYER0 + "ExternalEntity");
+ }
+
+ public void findBuiltins(ReadGraph g) throws DatabaseException {
+ RootLibrary = g.getBuiltin("http:/");
+ String = g.getBuiltin(CoreInitialization.LAYER0 + "String");
+ Library = g.getBuiltin(CoreInitialization.LAYER0 + "Library");
+ InstanceOf = g.getBuiltin(CoreInitialization.LAYER0 + "InstanceOf");
+ ConsistsOf = g.getBuiltin(CoreInitialization.LAYER0 + "ConsistsOf");
+ PartOf = g.getBuiltin(CoreInitialization.LAYER0 + "PartOf");
+ HasName = g.getBuiltin(CoreInitialization.LAYER0 + "HasName");
+ NameOf = g.getBuiltin(CoreInitialization.LAYER0 + "NameOf");
+ ExternalEntity = g.getBuiltin(CoreInitialization.LAYER0 + "ExternalEntity");
+ }
+
+ void addMissing(int handleIndex, String external) {
+ allMissingExternals.put(external, handleIndex);
+ Set<String> removals = new HashSet<>();
+ for(String ext : missingExternals) if(ext.startsWith(external + "/")) return;
+ for(String ext : missingExternals) if(external.startsWith(ext + "/")) removals.add(ext);
+ missingExternals.removeAll(removals);
+ missingExternals.add(external);
+ }
+
+ void prepare(ReadGraph graph) throws Exception {
+
+ Resource target = advisor.getTarget();
+ if(target != null)
+ indexRoot = graph.syncRequest(new PossibleIndexRoot(target));
+
+ findBuiltins(graph);
+ readIdentities(graph);
+
+// System.err.println("ext: " + extensions);
+// System.err.println("rc: " + resourceCount);
+// System.err.println("ic: " + identities.length);
+
+ ClusterBuilderFactory factory = graph.getService(ClusterBuilderFactory.class);
+ ClusterBuilder2 builder = factory.create(vg, false);
+
+ this.handles = new int[resourceCount];
+ TIntObjectMap<Identity> identityMap = TransferableGraphUtils.mapIdentities(identities);
+
+ // We must process roots first, because internal identifiers depend on them.
+ for(Identity identity : identities) {
+ IdentityDefinition definition = identity.definition;
+ if (definition instanceof Root) {
+ Root root = (Root) definition;
+ if (root.name.equals(""))
+ handles[identity.resource] = builder.handle(RootLibrary);
+ else {
+ Resource existing = advisor.analyzeRoot(graph, root);
+ if (existing != null)
+ handles[identity.resource] = builder.handle(existing);
+ }
+ }
+ }
+
+ for(Identity identity : identities) {
+ IdentityDefinition definition = identity.definition;
+ if(definition instanceof External) {
+ External def = (External)definition;
+ if(def.parent == -1) {
+ handles[identity.resource] = builder.handle(RootLibrary);
+ } else {
+ if("@inverse".equals(def.name)) {
+ int parent = handles[def.parent];
+ int child = builder.handle(graph.getInverse(builder.resource(parent)));
+ handles[identity.resource] = child;
+ } else {
+ int handle = handles[def.parent];
+ Resource parent = handle != 0 ? builder.resource(handle) : null;
+ // TODO: escape should be removed when names become well-behaving
+ if(parent != null) {
+ resolvedParents.put(graph.getURI(parent), parent);
+ Map<String,Resource> childMap = graph
+ .syncRequest(new UnescapedChildMapOfResource(parent),
+ TransientCacheAsyncListener.instance());
+ Resource child = childMap.get(def.name);
+ if(child == null) {
+ addMissing(identity.resource, graph.getURI(parent) + "/" + URIStringUtils.escape(def.name));
+ } else {
+ handles[identity.resource] = builder.handle(child);
+ }
+ } else {
+ addMissing(identity.resource, TransferableGraphUtils.getURI(resourceCount, identityMap, def.parent) + "/" + URIStringUtils.escape(def.name));
+ }
+ }
+ }
+ }
+ else if(definition instanceof Internal) {
+ String uri = TransferableGraphUtils.getURI(resourceCount, identityMap, identity.resource);
+ Resource existing = graph.getPossibleResource(uri);
+ if(existing != null) {
+ existingInternalMap.put(identity.resource, existing);
+ }
+ }
+ else if(definition instanceof Optional) {
+ External def = (External)definition;
+ Resource parent = builder.resource(handles[def.parent]);
+ if(parent != null)
+ handles[identity.resource] = builder.handle(graph.syncRequest(new UnescapedChildMapOfResource(parent)).get(def.name));
+ }
+ }
+
+ if (!missingExternals.isEmpty() && failOnMissingEntities())
+ throw new MissingDependencyException(this);
+ }
+
+ private boolean failOnMissingEntities() {
+ return "true".equalsIgnoreCase(
+ System.getProperty(
+ "org.simantics.tg.import.failOnMissingEntities",
+ "false") );
+ }
+
+ @Override
+ public Resource createChild(WriteOnlyGraph graph, Resource parent, Resource child, String name) throws DatabaseException {
+ //System.err.println("child " + parent + " - " + child + " = " + name);
+ if(child == null) child = graph.newResource();
+ Resource nameResource = graph.newResource();
+ graph.claim(nameResource, InstanceOf, null, String);
+ graph.claimValue(nameResource, name, WriteBindings.STRING);
+ graph.claim(child, HasName, NameOf, nameResource);
+ graph.claim(parent, ConsistsOf, PartOf, child);
+ return child;
+ }
+
+ int[] getClustering() {
+ if (vg != null)
+ return null;
+ Variant v = extensions.get(Extensions.CLUSTERING);
+ if(v == null) return null;
+ try {
+ return (int[])v.getValue(Bindings.INT_ARRAY);
+ } catch (AdaptException e) {
+ Logger.defaultLogError(e);
+ return null;
+ }
+ }
+
+ int[] getClusterSets() {
+ if (vg != null)
+ return null;
+ Variant v = extensions.get(Extensions.CLUSTER_SETS);
+ if(v == null) return null;
+ try {
+ return (int[])v.getValue(Bindings.INT_ARRAY);
+ } catch (AdaptException e) {
+ Logger.defaultLogError(e);
+ return null;
+ }
+ }
+
+ boolean needTranslation(Datatype type) {
+ return resourceUtil.mayHaveResource(type);
+ }
+
+ void findClusterSet(WriteOnlyGraph graph, Resource rootLibrary, int[] clustering, int[] clusterSets, long[] clusters, int id) throws DatabaseException {
+ ClusteringSupport support = graph.getService(ClusteringSupport.class);
+ if(id == Extensions.ROOT_LIBRARY_CLUSTER_SET || id == Extensions.INDEX_ROOT_CLUSTER_SET) return;
+ Resource indexRootClusterSetResource = rootLibrary;
+ if(indexRoot != null && support.isClusterSet(indexRoot)) {
+ indexRootClusterSetResource = indexRoot;
+ } else {
+ graph.setClusterSet4NewResource(rootLibrary);
+ graph.flushCluster();
+ }
+ int indexRootCsHandle = builder.handle(indexRootClusterSetResource);
+ for(int pos=0,index=0;index<clustering.length;index++) {
+ pos += clustering[index];
+ if(id < pos) {
+ int cs = clusterSets[index];
+ if(handles[id] == 0) {
+ int csHandle = 0;
+ if(cs == Extensions.ROOT_LIBRARY_CLUSTER_SET) csHandle = builder.handle(rootLibrary);
+ else if(cs == Extensions.INDEX_ROOT_CLUSTER_SET) {
+ if(indexRoot == null) throw new DatabaseException("No index root was available in TG import.");
+ csHandle = indexRootCsHandle;
+ }
+ else {
+ findClusterSet(graph, rootLibrary, clustering, clusterSets, clusters, cs);
+ csHandle = handles[cs];
+ }
+
+ if(clusters[index] != 0)
+ builder.selectCluster(clusters[index]);
+ else if(cs >= 0)
+ builder.newCluster(csHandle);
+
+ handles[id] = builder.newResource(csHandle);
+ clusters[index] = support.getCluster(builder.resource(handles[id]));
+
+ builder.createClusterSet(handles[id]);
+ }
+ return;
+ }
+ }
+ }
+
+ void createMissing(final WriteOnlyGraph graph) throws Exception {
+
+ if(allMissingExternals.isEmpty()) return;
+
+ XSupport xs = graph.getService(XSupport.class);
+ Pair<Boolean,Boolean> serviceMode = xs.getServiceMode();
+ xs.setServiceMode(true, false);
+ try {
+ ArrayList<String> missing = new ArrayList<>(allMissingExternals.keySet());
+ Collections.sort(missing);
+ for(String uri : missing) {
+ String[] parts = URIStringUtils.splitURI(uri);
+ // URIStringUtils.splitURI returns root URI in non-standard format, so fix it manually as a workaround
+ if (parts[0].equals("http://")) {
+ parts[0] = "http:/";
+ }
+
+ Resource parent = resolvedParents.get(parts[0]);
+ // TODO: proper exception message
+ if(parent == null) {
+ throw new IllegalStateException("Missing URI: " + uri);
+ }
+
+ Resource childResource = graph.newResource();
+ graph.claim(childResource, InstanceOf, null, ExternalEntity);
+
+ Resource nameResource = graph.newResource();
+ graph.claim(nameResource, InstanceOf, null, String);
+ graph.claimValue(nameResource, URIStringUtils.unescape(parts[1]), WriteBindings.STRING);
+ graph.claim(childResource, HasName, NameOf, nameResource);
+
+ graph.claim(parent, ConsistsOf, PartOf, childResource);
+
+ resolvedParents.put(uri, childResource);
+
+ handles[allMissingExternals.get(uri)] = builder.handle(childResource);
+ }
+ } finally {
+ xs.setServiceMode(serviceMode.first, serviceMode.second);
+ }
+ }
+
+ void write(final WriteOnlyGraph graph) throws Exception {
+
+ final SerialisationSupport ss = graph.getService(SerialisationSupport.class);
+
+ ClusterBuilderFactory factory = graph.getService(ClusterBuilderFactory.class);
+ if(advisor instanceof IImportAdvisor2) {
+ boolean allowImmutable = ((IImportAdvisor2)advisor).allowImmutableModifications();
+ builder = factory.create(vg, allowImmutable);
+ } else {
+ builder = factory.create(vg, false);
+ }
+
+ createMissing(graph);
+
+ final int[] handles = this.handles;
+
+ int[] clustering = getClustering();
+ if(clustering != null) {
+
+ int[] clusterSets = getClusterSets();
+ if(clusterSets != null) {
+
+ assert(clustering.length == clusterSets.length);
+
+ long[] clusters = new long[clustering.length];
+
+ // Create clustering
+ for(int i=0;i<clusterSets.length;i++) {
+ findClusterSet(graph, graph.getRootLibrary(), clustering, clusterSets, clusters, clusterSets[i]);
+ }
+
+ // Then create all resources
+ int i=0;
+ for(int j=0;j<clustering.length;j++) {
+ int c = clustering[j];
+ int s = clusterSets[j];
+ int setHandle = 0;
+ if(s == Extensions.ROOT_LIBRARY_CLUSTER_SET)
+ setHandle = builder.handle(graph.getRootLibrary());
+ else if(s == Extensions.INDEX_ROOT_CLUSTER_SET) {
+ if(indexRoot == null) throw new DatabaseException("No index root was available in TG import.");
+ setHandle = builder.handle(indexRoot);
+ }
+ else setHandle = handles[s];
+ // Preserve clustering only for internal resources
+ if(clusters[j] != 0)
+ builder.selectCluster(clusters[j]);
+ else if(s >= 0)
+ builder.newCluster(setHandle);
+ for(int r=0;r<c;r++, i++)
+ if(handles[i] == 0) handles[i] = builder.newResource();
+ }
+
+ for(;i<handles.length;++i)
+ if(handles[i] == 0) handles[i] = builder.newResource();
+
+ } else {
+
+ int i = 0;
+ for(int c : clustering) {
+ builder.newCluster();
+ for(int r=0;r<c;r++, i++)
+ if(handles[i] == 0) handles[i] = builder.newResource();
+ }
+
+ for(;i<handles.length;++i)
+ if(handles[i] == 0) handles[i] = builder.newResource();
+
+ }
+
+ } else {
+
+ // Create blank resources
+ for(int i=0;i<handles.length;++i)
+ if(handles[i] == 0) handles[i] = builder.newResource();
+
+ }
+
+ // Internal identities
+ for(Identity identity : identities) {
+ IdentityDefinition definition = identity.definition;
+// if(handles[identity.resource] != 0)
+// continue;
+ if(definition instanceof External) {
+ // Already done everything
+ }
+ else if(definition instanceof Internal) {
+ Internal def = (Internal)definition;
+ Resource external = existingInternalMap.get(identity.resource);
+ if(external != null) {
+ handles[identity.resource] = builder.handle(external);
+ } else {
+ if(handles[identity.resource] != 0)
+ handles[identity.resource] = builder.handle(advisor.createChild(graph, this, builder.resource(handles[def.parent]), builder.resource(handles[identity.resource]), def.name));
+ else
+ handles[identity.resource] = builder.handle(advisor.createChild(graph, this, builder.resource(handles[def.parent]), null, def.name));
+ }
+
+ }
+ else if(definition instanceof Root) {
+
+ Root root = (Root)definition;
+ if(handles[identity.resource] != 0)
+ handles[identity.resource] = builder.handle(advisor.createRoot(graph, root, builder.resource(handles[identity.resource])));
+ else
+ handles[identity.resource] = builder.handle(advisor.createRoot(graph, root, null));
+ }
+ else if(definition instanceof Optional) {
+ Optional def = (Optional)definition;
+ if(handles[identity.resource] != 0) {
+ Resource child = advisor.createChild(graph, this, builder.resource(handles[def.parent]), builder.resource(handles[identity.resource]), def.name);
+ graph.claim(child, InstanceOf, null, Library); // ???
+ handles[identity.resource] = builder.handle(child);
+ } else {
+ Resource child = advisor.createChild(graph, this, builder.resource(handles[def.parent]), null, def.name);
+ graph.claim(child, InstanceOf, null, Library); // ???
+ handles[identity.resource] = builder.handle(child);
+ }
+ }
+ }
+
+ int[] done = { 0 };
+ int[] percentage = { 0 };
+
+ int statementCount = tg.getStatementCount();
+ tg.forStatements(null, new TransferableGraphSourceProcedure<int[]>() {
+
+ @Override
+ public void execute(int[] value) throws Exception {
+
+ int sub = value[0];
+ int pred = value[1];
+ int inv = value[2];
+ int obj = value[3];
+
+ int subject = handles[sub];
+ int predicate = handles[pred];
+ int object = handles[obj];
+
+ builder.addStatement(graph, subject, predicate, object);
+ if(inv >= 0) {
+ int inverse = handles[inv];
+ builder.addStatement(graph, object, inverse, subject);
+ }
+
+ // Count from 0% -> 50% => total = statementCount*2
+ percentage[0] = updatePercentage(percentage[0], done[0]++, statementCount*2);
+
+ }
+
+ });
+
+ int valueCount = tg.getValueCount();
+ done[0] = 0;
+
+ class ValueProcedure extends InputStream implements TransferableGraphSourceValueProcedure {
+
+ private TGResourceUtil util = new TGResourceUtil();
+ private DataInput source;
+
+ @Override
+ public void execute(int _resource, Datatype type, DataInput stream) throws Exception {
+
+ source = stream;
+
+ //int file = _resource & 0x80000000;
+ int resource = _resource & 0x7FFFFFFF;
+
+ Binding binding = Bindings.getBinding(type);
+ Serializer s = Bindings.getSerializer(binding);
+
+ builder.beginValue(handles[resource]);
+ if(util.mayHaveResource(type)) {
+ Object value = s.deserialize(stream);
+ util.adaptValue( binding, value, new LongAdapter() {
+ @Override
+ public long adapt(long in) {
+ try {
+ return ss.getRandomAccessId(handles[(int)in]);
+ } catch (DatabaseException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+ });
+ byte[] bytes = s.serialize(value);
+ for(byte b : bytes) {
+ int val = b;
+ if(val < 0) val += 256;
+ builder.appendValue(val);
+ }
+ } else {
+ s.skip(this);
+ }
+ builder.endValue();
+ work();
+
+ }
+
+ @Override
+ public int read() throws IOException {
+ int value = source.readUnsignedByte();
+ try {
+ builder.appendValue(value);
+ } catch (DatabaseException e) {
+ LOGGER.error("Failed to write value into database", e);
+ }
+ return value;
+ }
+
+ @Override
+ public void rawCopy(int resource, int length, DataInput input) throws Exception {
+ builder.beginValue(handles[resource]);
+ for (int i = 0; i < length; ++i)
+ builder.appendValue(input.readUnsignedByte());
+ builder.endValue();
+ work();
+ }
+
+ private void work() {
+ // Count from 50% -> 100% => [valueCount, valueCount*2)
+ percentage[0] = updatePercentage(percentage[0], valueCount + done[0]++, valueCount*2);
+ }
+ };
+
+ tg.forValues2(null, new ValueProcedure());
+
+ for(Resource r : existingInternalMap.valueCollection()) {
+ try {
+ graph.deny(r, InstanceOf, null, ExternalEntity, null);
+ } catch (DatabaseException e) {
+ graph.deny(r, InstanceOf, null, ExternalEntity, vg);
+ }
+ }
+
+ }
+
+ @Override
+ public long[] getResourceIds(SerialisationSupport serializer) throws DatabaseException {
+ final int count = handles.length;
+ long[] resourceIds = new long[count];
+ for(int i=0;i<count;++i)
+ resourceIds[i] = serializer.getRandomAccessId(handles[i]);
+ return resourceIds;
+ }
}
\ No newline at end of file
Code Review / simantics / platform.git / blobdiff